Merge pull request #103

Rewrite Zeek plugin to convert to/from STIX-2 Indicators and Sightings

CHANGELOG.md

@@ -12,14 +12,29 @@ Every entry has a category for which we use the following visual abbreviations:
 
 ## Unreleased
 
+- 🐞 We fixed a bug in the routing logic for SnapshotRequests. Apps can now
+  request snapshots as expected for all `stix2`-prefixed topics.
+  [#103](https://github.com/tenzir/threatbus/pull/103)
+
+- 🎁 The Zeek plugin now supports the
+  [STIX-2 (version 2.1)](https://docs.oasis-open.org/cti/stix/v2.1/stix-v2.1.html)
+  standard for
+  [Indicators](https://docs.oasis-open.org/cti/stix/v2.1/cs02/stix-v2.1-cs02.html#_muftrcpnf89v)
+  and [Sightings](https://docs.oasis-open.org/cti/stix/v2.1/cs02/stix-v2.1-cs02.html#_a795guqsap3r).
+  The plugin converts STIX-2 Indicators on best-effort basis to
+  [Zeek Intel items](https://docs.zeek.org/en/current/scripts/base/frameworks/intel/main.zeek.html#type-Intel::Type)
+  before forwarding them to Zeek. Likewise, the plugin converts Zeek sightings
+  to valid STIX-2 Sightings before publishing them on Threat Bus topics.
+  [#103](https://github.com/tenzir/threatbus/pull/103)
+
 - 🎁 The MISP plugin now supports the
   [STIX-2 (version 2.1)](https://docs.oasis-open.org/cti/stix/v2.1/stix-v2.1.html)
   standard for
   [Indicators](https://docs.oasis-open.org/cti/stix/v2.1/cs02/stix-v2.1-cs02.html#_muftrcpnf89v)
   and [Sightings](https://docs.oasis-open.org/cti/stix/v2.1/cs02/stix-v2.1-cs02.html#_a795guqsap3r).
   The plugin converts MISP attributes to valid STIX-2 Indicators on best-effort
   basis before publishing them on Threat Bus topics. Likewise, the plugin
-  converts STIX-2 sightings to MISP sightings before sending them the MISP.
+  converts STIX-2 Sightings to MISP sightings before sending them the MISP.
   [#102](https://github.com/tenzir/threatbus/pull/102)
 
 - 🐞 We fixed a bug in the JSON (de-)serialization logic for `SnapshotEnvelope`s

Dockerfile

@@ -2,9 +2,6 @@ FROM fixel/zeek:broker-latest
 
 RUN apt-get -qq update && apt-get -qqy install \
   python3-pip wget software-properties-common gnupg2
-RUN wget -qO - https://packages.confluent.io/deb/5.4/archive.key | apt-key add - && \
-  add-apt-repository "deb [arch=amd64] https://packages.confluent.io/deb/5.4 stable main" && \
-  apt-get update -qqy && apt-get install -qqy confluent-platform-2.12
 
 RUN pip3 install --upgrade pip
 
@@ -16,9 +13,9 @@ COPY README.md .
 COPY threatbus threatbus
 COPY plugins plugins
 RUN python3 -m pip install . && \
-  #python3 -m pip install plugins/apps/threatbus_misp && \
-  #python3 -m pip install plugins/apps/threatbus_zeek && \
   #python3 -m pip install plugins/apps/threatbus_cif3 && \
+  python3 -m pip install plugins/apps/threatbus_misp[zmq] && \
+  python3 -m pip install plugins/apps/threatbus_zeek && \
   python3 -m pip install plugins/apps/threatbus_zmq_app && \
   python3 -m pip install plugins/backbones/threatbus_inmem && \
   python3 -m pip install plugins/backbones/threatbus_rabbitmq

Makefile

@@ -22,10 +22,10 @@ unit-tests:
 	$(MAKE) -C plugins/backbones/threatbus_rabbitmq unit-tests
 	$(MAKE) -C plugins/apps/threatbus_zmq_app unit-tests
 	$(MAKE) -C plugins/apps/threatbus_misp unit-tests
+	$(MAKE) -C plugins/apps/threatbus_zeek unit-tests
   # Threat Bus is currently being migrated to use STIX-2 as internal format.
   # For the time being, all un-migrated plugins cannot be not tested against the
   # current master
-	#$(MAKE) -C plugins/apps/threatbus_zeek unit-tests
 	#$(MAKE) -C plugins/apps/threatbus_cif3 unit-tests
 	#$(MAKE) -C apps/vast unit-tests
 
@@ -36,9 +36,10 @@ integration-tests:
 	docker run -d --rm --hostname=test-rabbit --name=rabbit-int -p 35672$(:)5672 rabbitmq$(:)3 > /dev/null 2>&1
 	-python -m unittest tests/integration/test_misp_inmem.py
 	-python -m unittest tests/integration/test_zmq_app_management.py
+	-python -m unittest tests/integration/test_zmq_app_message_roundtrips.py
+	-python -m unittest tests/integration/test_zeek_app.py
+	-python -m unittest tests/integration/test_zeek_inmem.py
 	# -python -m unittest tests/integration/test_rabbitmq.py
-	# -python -m unittest tests/integration/test_zeek_app.py
-	# -python -m unittest tests/integration/test_message_roundtrips.py
 	-${RM} {broker,intel,reporter,weird}.log
 	docker kill rabbit-int > /dev/null 2>&1
 
@@ -93,9 +94,9 @@ dev-mode:
 	$(MAKE) -C plugins/backbones/threatbus_inmem dev-mode
 	$(MAKE) -C plugins/backbones/threatbus_rabbitmq dev-mode
 	$(MAKE) -C plugins/apps/threatbus_misp dev-mode
+	$(MAKE) -C plugins/apps/threatbus_zeek dev-mode
   # Threat Bus is currently being migrated to use STIX-2 as internal format.
   # For the time being, all un-migrated plugins cannot be run in conjunction
   # with current master
-	# $(MAKE) -C plugins/apps/threatbus_zeek dev-mode
 	# $(MAKE) -C plugins/apps/threatbus_cif3 dev-mode
 	# $(MAKE) -C apps/vast dev-mode

README.md

@@ -60,7 +60,7 @@ Threat Bus. There are more integrations available, so make sure to check out all
 
 ```sh
 mv config.yaml.example config.yaml   # rename example config file
-venv/bin/threatbus -c config.yaml
+threatbus -c config.yaml
 ```
 
 *Start Zeek as Threat Bus app*
@@ -72,7 +72,7 @@ zeek -i <INTERFACE> -C ./apps/zeek/threatbus.zeek
 *Start Zeek and request a snapshot*
 
 ```sh
-zeek -i <INTERFACE> -C ./apps/zeek/threatbus.zeek -- "Tenzir::snapshot_intel=30 days"
+zeek -i <INTERFACE> -C ./apps/zeek/threatbus.zeek "Tenzir::snapshot_intel=30 days"
 ```
 
 Threat Bus also ships as pre-built Docker image and is available on
@@ -101,7 +101,7 @@ source venv/bin/activate              # optional
 pip install threatbus
 pip install threatbus-inmem           # inmemory backbone plugin
 pip install threatbus-rabbitmq        # RabbitMQ backbone plugin
-pip install threatbus-misp            # MISP application plugin
+pip install threatbus-misp[zmq]       # MISP application plugin
 pip install threatbus-zeek            # Zeek application plugin
 pip install threatbus-<plugin_name>
 ```

apps/zeek/threatbus.zeek

@@ -9,20 +9,22 @@
 module Tenzir;
 
 export {
-  ## Threat Bus representation of intelligence.
+  ## Threat Bus representation of intelligence items (IoCs).
   type Intelligence: record {
     ## A timestamp
     ts: time;
     ## A unique identifier for the intel item.
     id: string;
     ## The intel type according to Intel::Type.
-    data: table[string] of string;
+    intel_type: string;
+    ## The IoC value, e.g., "evil.com"
+    indicator: string;
     ## The operation to perform (either ADD or REMOVE)
     operation: string;
   };
 
   ## Broker bind address.
-  option broker_host = "localhost" &redef;
+  option broker_host = "127.0.0.1" &redef;
 
   ## Broker port.
   option broker_port = 47761/tcp &redef;
@@ -43,15 +45,15 @@ export {
   option log_operations = T &redef;
 
   ## Threat Bus topic to subscribe to for receiving intel.
-  option intel_topic = "threatbus/intel" &redef;
+  option intel_topic = "stix2/indicator" &redef;
 
   ## Event to raise for intel item insertion.
   ##
   ## item: The intel type to add.
   global intel: event(item: Intelligence);
 
   ## Threat Bus topic to report sightings.
-  option sighting_topic = "threatbus/sighting" &redef;
+  option sighting_topic = "stix2/sighting" &redef;
 
   ## Event to report back sightings of (previously added) intel.
   ##
@@ -94,20 +96,52 @@ export {
 
 ## ---------- broker management and logging ------------------------------------
 
+# Predicate for checking if Zeek is already subscribed to Threat Bus
+function is_subscribed(): bool
+  {
+  return p2p_topic != "";
+  }
+
+# Unsubscribe from the Threat Bus p2p_topic by sending an `Unsubscribe` event
+# via Broker to the Threat Bus Zeek plugin's management endpoint.
+function unsubscribe_p2p()
+  {
+  if ( log_operations )
+    Reporter::info(fmt("unsubscribing from p2p_topic %s", p2p_topic));
+  Broker::publish(management_topic, unsubscribe, p2p_topic);
+  p2p_topic = ""; # invalidate old p2p_topic
+  }
+
+# Subscribe to Threat Bus by sending a `Subscribe` event via Broker to the
+# Threat Bus Zeek plugin's management endpoint.
+function subscribe_p2p()
+  {
+  Broker::publish(management_topic, subscribe, intel_topic, snapshot_intel);
+  }
+
 event Broker::peer_lost(endpoint: Broker::EndpointInfo, msg: string)
   {
   if ( endpoint?$network && endpoint$network$address == broker_host 
       && endpoint$network$bound_port == broker_port && log_operations )
-    Reporter::info("threatbus unpeered");
+    Reporter::info("Threat Bus unpeered");
   }
 
 event Broker::peer_added(endpoint: Broker::EndpointInfo, msg: string)
   {
   if ( endpoint?$network && endpoint$network$address == broker_host 
       && endpoint$network$bound_port == broker_port )
     if ( log_operations )
-      Reporter::info("threatbus peered");
-    Broker::publish(management_topic, subscribe, intel_topic, snapshot_intel);
+      Reporter::info("Threat Bus peered");
+    if ( is_subscribed() )
+      {
+      # Already peered, so Zeek has lost connection to Threat Bus. The reason
+      # for the connection loss is unclear. If the Threat Bus host has died we
+      # need to invalidate our old p2p_topic. If it has not, and we invalidate
+      # the p2p_topic, Threat Bus will be left with a dangling subscription.
+      # So we unsubsribe our old p2p_topic entirely and re-subscribe.
+      unsubscribe_p2p();
+      }
+    subscribe_p2p();
   }
 
 # Only the manager communicates with Threat Bus.
@@ -117,7 +151,8 @@ event zeek_init() &priority=1
   {
   if ( log_operations )
     {
-    Reporter::info(fmt("subscribing to management topic %s", management_topic));
+    Reporter::info(fmt("subscribing to management topic %s with snapshot request for %s",
+                       management_topic, snapshot_intel));
     Reporter::info(fmt("reporting noisy intel at %d matches/sec",
                        noisy_intel_threshold));
     }
@@ -132,7 +167,7 @@ event zeek_init() &priority=1
 event zeek_init() &priority=0
   {
   if ( log_operations )
-    Reporter::info(fmt("peering with threatbus at %s:%s",
+    Reporter::info(fmt("peering with Threat Bus at %s:%s",
                        broker_host, broker_port));
   Broker::peer(broker_host, broker_port, 5sec);
   }
@@ -143,9 +178,7 @@ event zeek_init() &priority=0
       || Cluster::local_node_type() == Cluster::MANAGER )
 event zeek_done() &priority=1
   {
-  Broker::publish(management_topic, unsubscribe, p2p_topic);
-  if ( log_operations )
-    Reporter::info(fmt("unsubscribing from p2p_topic %s", p2p_topic));
+  unsubscribe_p2p();
   }
 @endif
 
@@ -177,8 +210,8 @@ global intel_type_map: table[string] of Intel::Type = {
 function map_to_zeek_intel(item: Intelligence): Intel::Item
   {
   local intel_item: Intel::Item = [
-    $indicator = item$data["indicator"],
-    $indicator_type = intel_type_map[item$data["intel_type"]],
+    $indicator = item$indicator,
+    $indicator_type = intel_type_map[item$intel_type],
     $meta = record(
       $desc = item$id,
       $url = tb_intel_tag, # re-used to identify Threat Bus as sending entity
@@ -191,10 +224,10 @@ function map_to_zeek_intel(item: Intelligence): Intel::Item
 
 function is_mappable_intel(item: Intelligence): bool
   {
-  return item?$data && "indicator" in item$data && "intel_type" in item$data && item$data["intel_type"] in intel_type_map;
+  return item?$indicator && item?$intel_type && item$intel_type in intel_type_map;
   }
 
-# Event sent by Threat Bus to indicate a change of known intelligence to Zeek.
+# Event sent by Threat Bus to publish a new intelligence item (IoC) to Zeek.
 event intel(item: Intelligence)
   {
   if ( ! is_mappable_intel(item) ) {
@@ -214,8 +247,13 @@ event intel(item: Intelligence)
 # Event sent by Threat Bus to acknowledge new subscriptions.
 event subscription_acknowledged(topic: string)
   {
-  # This particular topic is used by Threat Bus to send dedicated messages to
-  # only this Zeek instance.
+  # Avoid picking up p2p_topics from other Zeek instances. This is a shortcoming
+  # of Broker, as we cannot avoid seeing all messages on the management topic. 
+  if ( is_subscribed() )
+    return;
+
+  # This particular topic is used by Threat Bus to send peer-to-peer messages to
+  # this Zeek instance only.
   p2p_topic = topic;
   Broker::subscribe(p2p_topic);
   if ( log_operations )
@@ -236,12 +274,12 @@ event Intel::match(seen: Intel::Seen, items: set[Intel::Item])
       next;
     if ( ! item$meta?$desc )
       {
-      Reporter::error("skipping threatbus intel item without ID");
+      Reporter::error("skipping Threat Bus intel item without ID");
       next;
       }
     local intel_id = item$meta$desc;
     if ( log_operations )
-      Reporter::info(fmt("sighted threatbus intel with ID: %s", intel_id));
+      Reporter::info(fmt("sighted Threat Bus intel with ID: %s", intel_id));
     local n = ++sightings[intel_id];
     local noisy = noisy_intel_threshold != 0 && n > noisy_intel_threshold;
     local context: table[string] of any;

plugins/apps/threatbus_zeek/setup.py

@@ -26,7 +26,9 @@
     description="A plugin to enable threatbus communication with Zeek network monitor.",
     entry_points={"threatbus.app": ["zeek = threatbus_zeek.plugin"]},
     install_requires=[
-        "threatbus >= 2020.12.16, < 2021.2.24",
+        "stix2 >= 2.1",
+        "stix2-patterns >= 0.3",
+        "threatbus >= 2021.2.24",
     ],
     keywords=[
         "Zeek",