Improve message distribution performance

[0snap]

📔 Description

Improve (retro-matching) performance.

VAST Bridge:

• use asynchronous tasks to query VAST concurrently
• add configurable ulimit to control task count

Threat Bus:

• use multiprocessing.JoinableQueue for message passing
• use select system call (possible via JoinableQueue) to speed up message distribution
• add StoppableWorker base-class for background threads
• control shutdown behavior of plugins

📝 Checklist

• All user-facing changes have changelog entries.
• The changes are reflected on docs.tenzir.com/vast, if necessary.
• The PR description contains instructions for the reviewer, if necessary.

🎯 Review Instructions

File-by-file.

Test this as follows:

• setup a virtual environment, somewhere on your computer, via virtualenv venv
• activate it source venv/bin/activate
• run make dev-mode
• write a config that activates all plugins (install broker, CIF, MISP, RabbitMQ ... locally first, see READMEs)
• run Threat Bus via venv/bin/threatbus -c config.yaml
• Send around IoCs: create / delete in MISP, verify they make it to Zeek / VAST, browse sites that trigger these IoCs, watch the sightings being collected and sent back.

[lgtm-com]

This pull request introduces 1 alert when merging 996d3ef into 4ddca09 - view on LGTM.com

new alerts:

• 1 for Unused import

[lgtm-com]

This pull request introduces 1 alert when merging ade0559 into 4ddca09 - view on LGTM.com

new alerts:

• 1 for Unused import

[lgtm-com]

This pull request introduces 1 alert when merging a361c63 into 4ddca09 - view on LGTM.com

new alerts:

• 1 for Unused import

[0snap]

@mavam I'll make a separate PR for the docs update later this week.

[0snap]

@mavam I updated the PR description with a recipe for testing everything interactively.

[mavam]

I'm having trouble with make dev-mode, which attempts to build a bunch of dependencies but fails somewhere in the middle with Python 3.8 errors:

...
/usr/local/Cellar/python@3.8/3.8.6_1/Frameworks/Python.framework/Versions/3.8/include/python3.8/cpython/pystate.h:81:15: note:
      'curexc_traceback' declared here
    PyObject *curexc_traceback;
              ^
msgpack/_packer.cpp:5753:24: error: no member named 'exc_type' in '_ts'
    tmp_type = tstate->exc_type;
               ~~~~~~  ^
msgpack/_packer.cpp:5754:25: error: no member named 'exc_value' in '_ts'; did
      you mean 'curexc_value'?
    tmp_value = tstate->exc_value;
                        ^~~~~~~~~
                        curexc_value
/usr/local/Cellar/python@3.8/3.8.6_1/Frameworks/Python.framework/Versions/3.8/include/python3.8/cpython/pystate.h:80:15: note:
      'curexc_value' declared here
    PyObject *curexc_value;
              ^
msgpack/_packer.cpp:5755:22: error: no member named 'exc_traceback' in '_ts';
      did you mean 'curexc_traceback'?
    tmp_tb = tstate->exc_traceback;
                     ^~~~~~~~~~~~~
                     curexc_traceback
...

However, the make run exited with code 0.

The bigger issue is that I need a bit of handholding for getting the proper dev setup locally going. It'd be nice if we had something where we can call docker-compose and get Zeek, MISP, Kafka, and all other moving parts in one shot so that all that's left is Threat Bus testing.

@mavam I rebased this branch on current master and quantified the improvements in the CHANGELOGs. Can you give this another spin, please?

[mavam]

Here's how I tested this:

# Terminal 1
python3 -m venv venv
source venv/bin/activate
python -m venv --system-site-packages venv
make dev-mode
threatbus -c config.yaml

# Terminal 2
./pyvast_threatbus.py -c config.yaml

# Terminal 3
vast start

# Terminal 3
python /tmp/rabbitmq_sender.py

At first, I had a transform-context in there, but forgot to install fever, so the bridge crashed hard:

Traceback (most recent call last):
  File "./pyvast_threatbus.py", line 669, in <module>
    main()
  File "./pyvast_threatbus.py", line 650, in main
    asyncio.run(
  File "/usr/local/Cellar/python@3.8/3.8.6_1/Frameworks/Python.framework/Versions/3.8/lib/python3.8/asyncio/runners.py", line 44, in run
    return loop.run_until_complete(main)
  File "/usr/local/Cellar/python@3.8/3.8.6_1/Frameworks/Python.framework/Versions/3.8/lib/python3.8/asyncio/base_events.py", line 616, in run_until_complete
    return future.result()
  File "./pyvast_threatbus.py", line 170, in start
    return await asyncio.gather(*async_tasks)
  File "./pyvast_threatbus.py", line 441, in report_sightings
    context_str = await invoke_cmd_for_context(
  File "./pyvast_threatbus.py", line 386, in invoke_cmd_for_context
    proc = await asyncio.create_subprocess_exec(
  File "/usr/local/Cellar/python@3.8/3.8.6_1/Frameworks/Python.framework/Versions/3.8/lib/python3.8/asyncio/subprocess.py", line 236, in create_subprocess_exec
    transport, protocol = await loop.subprocess_exec(
  File "/usr/local/Cellar/python@3.8/3.8.6_1/Frameworks/Python.framework/Versions/3.8/lib/python3.8/asyncio/base_events.py", line 1630, in subprocess_exec
    transport = await self._make_subprocess_transport(
  File "/usr/local/Cellar/python@3.8/3.8.6_1/Frameworks/Python.framework/Versions/3.8/lib/python3.8/asyncio/unix_events.py", line 197, in _make_subprocess_transport
    transp = _UnixSubprocessTransport(self, protocol, args, shell,
  File "/usr/local/Cellar/python@3.8/3.8.6_1/Frameworks/Python.framework/Versions/3.8/lib/python3.8/asyncio/base_subprocess.py", line 36, in __init__
    self._start(args=args, shell=shell, stdin=stdin, stdout=stdout,
  File "/usr/local/Cellar/python@3.8/3.8.6_1/Frameworks/Python.framework/Versions/3.8/lib/python3.8/asyncio/unix_events.py", line 789, in _start
    self._proc = subprocess.Popen(
  File "/usr/local/Cellar/python@3.8/3.8.6_1/Frameworks/Python.framework/Versions/3.8/lib/python3.8/subprocess.py", line 854, in __init__
    self._execute_child(args, executable, preexec_fn, close_fds,
  File "/usr/local/Cellar/python@3.8/3.8.6_1/Frameworks/Python.framework/Versions/3.8/lib/python3.8/subprocess.py", line 1702, in _execute_child
    raise child_exception_type(errno_num, err_msg, err_filename)
FileNotFoundError: [Errno 2] No such file or directory: 'fever'
2020-11-17 18:03:08 INFO     Unsubscribing from topic 'rcdzkdrjngahzxuoabvcgvzjoautwnfx' ...
2020-11-17 18:03:08 INFO     Unsubscription successful

The I commented the transformation and could feed data to VAST via the rabbitmq_sender.py helper from your benchmark branch. This produced a bunch of results:

{"timestamp": "2009-11-19T09:56:19.842454", "flow_id": 1899585223580261, "pcap_cnt": null, "src_ip": "192.168.1.103", "src_port": 1065, "dest_ip": "68.216.79.113", "dest_port": 37, "proto": "TCP", "event_type": "flow", "community_id": "1:txwUbnSdHQMSYgtcdEaq+0xhvDA=", "flow": {"pkts_toserver": 6, "pkts_toclient": 0, "bytes_toserver": 396, "bytes_toclient": 0, "start": "2009-11-19T03:59:20.645733", "end": "2009-11-19T03:59:29.655330", "age": 9, "state": "new", "reason": "timeout", "alerted": false}, "app_proto": null, "source": "VAST"}
2020-11-17 18:05:47 DEBUG    Reported sighting: Sighting(ts=datetime.datetime(2009, 11, 19, 9, 56, 19, 842454), intel='intel-42', context={'timestamp': '2009-11-19T09:56:19.842454', 'flow_id': 1899585223580261, 'pcap_cnt': None, 'src_ip': '192.168.1.103', 'src_port': 1065, 'dest_ip': '68.216.79.113', 'dest_port': 37, 'proto': 'TCP', 'event_type': 'flow', 'community_id': '1:txwUbnSdHQMSYgtcdEaq+0xhvDA=', 'flow': {'pkts_toserver': 6, 'pkts_toclient': 0, 'bytes_toserver': 396, 'bytes_toclient': 0, 'start': '2009-11-19T03:59:20.645733', 'end': '2009-11-19T03:59:29.655330', 'age': 9, 'state': 'new', 'reason': 'timeout', 'alerted': False}, 'app_proto': None, 'source': 'VAST'}, ioc=('68.216.79.113',))
{"timestamp": "2009-11-19T09:56:19.842454", "flow_id": 1360622762622986, "pcap_cnt": null, "src_ip": "192.168.1.103", "src_port": 1076, "dest_ip": "68.216.79.113", "dest_port": 37, "proto": "TCP", "event_type": "flow", "community_id": "1:DKcIHk6yBoGg40TfT9kKwdmXBnw=", "flow": {"pkts_toserver": 6, "pkts_toclient": 0, "bytes_toserver": 396, "bytes_toclient": 0, "start": "2009-11-19T04:24:48.884746", "end": "2009-11-19T04:24:57.770677", "age": 9, "state": "new", "reason": "timeout", "alerted": false}, "app_proto": null, "source": "VAST"}

So my brief test seems to work. I'm going to review the code separately later.