implement dynamic modules

cmd/pktvisor-pcap/main.cpp

@@ -82,7 +82,7 @@ int main(int argc, char *argv[])
         logger->set_level(spdlog::level::debug);
     }
 
-    CoreRegistry mgrs(nullptr);
+    CoreRegistry registry;
 
     std::signal(SIGINT, signal_handler);
     std::signal(SIGTERM, signal_handler);
@@ -125,8 +125,8 @@ int main(int argc, char *argv[])
         input_stream->info_json(j["info"]);
         logger->info("{}", j.dump(4));
 
-        mgrs.input_manager()->module_add(std::move(input_stream));
-        auto [input_stream_, stream_mgr_lock] = mgrs.input_manager()->module_get_locked("pcap");
+        registry.input_manager()->module_add(std::move(input_stream));
+        auto [input_stream_, stream_mgr_lock] = registry.input_manager()->module_get_locked("pcap");
         stream_mgr_lock.unlock();
         auto pcap_stream = dynamic_cast<input::pcap::PcapInputStream *>(input_stream_);
 
@@ -140,8 +140,8 @@ int main(int argc, char *argv[])
             auto handler_module = std::make_unique<handler::net::NetStreamHandler>("net", pcap_stream, &window_config);
             handler_module->config_set("recorded_stream", true);
             handler_module->start();
-            mgrs.handler_manager()->module_add(std::move(handler_module));
-            auto [handler, handler_mgr_lock] = mgrs.handler_manager()->module_get_locked("net");
+            registry.handler_manager()->module_add(std::move(handler_module));
+            auto [handler, handler_mgr_lock] = registry.handler_manager()->module_get_locked("net");
             handler_mgr_lock.unlock();
             net_handler = dynamic_cast<handler::net::NetStreamHandler *>(handler);
         }
@@ -150,8 +150,8 @@ int main(int argc, char *argv[])
             auto handler_module = std::make_unique<handler::dns::DnsStreamHandler>("dns", pcap_stream, &window_config);
             handler_module->config_set("recorded_stream", true);
             handler_module->start();
-            mgrs.handler_manager()->module_add(std::move(handler_module));
-            auto [handler, handler_mgr_lock] = mgrs.handler_manager()->module_get_locked("dns");
+            registry.handler_manager()->module_add(std::move(handler_module));
+            auto [handler, handler_mgr_lock] = registry.handler_manager()->module_get_locked("dns");
             handler_mgr_lock.unlock();
             dns_handler = dynamic_cast<handler::dns::DnsStreamHandler *>(handler);
         }
@@ -160,8 +160,8 @@ int main(int argc, char *argv[])
             auto handler_module = std::make_unique<handler::dhcp::DhcpStreamHandler>("dhcp", pcap_stream, &window_config);
             handler_module->config_set("recorded_stream", true);
             handler_module->start();
-            mgrs.handler_manager()->module_add(std::move(handler_module));
-            auto [handler, handler_mgr_lock] = mgrs.handler_manager()->module_get_locked("dhcp");
+            registry.handler_manager()->module_add(std::move(handler_module));
+            auto [handler, handler_mgr_lock] = registry.handler_manager()->module_get_locked("dhcp");
             handler_mgr_lock.unlock();
             dhcp_handler = dynamic_cast<handler::dhcp::DhcpStreamHandler *>(handler);
         }

cmd/pktvisord/main.cpp

@@ -12,6 +12,7 @@
 #include "handlers/static_plugins.h"
 #include "inputs/static_plugins.h"
 #include "visor_config.h"
+#include <Corrade/Utility/ConfigurationGroup.h>
 #include <docopt/docopt.h>
 #include <resolv.h>
 #include <spdlog/sinks/basic_file_sink.h>
@@ -32,9 +33,9 @@ static const char USAGE[] =
 
     pktvisord summarizes data streams and exposes a REST API control plane for configuration and metrics.
 
-    pktvisord operation is configured via Taps and Collection Policies. The former set up the available
-    input streams while the latter instantiate Taps and Stream Handlers to analyze and summarize
-    the stream data.
+    pktvisord operation is configured via Taps and Collection Policies. Taps abstract the process of "tapping into"
+    input streams with templated configuration while Policies use Taps to instantiate and configure Input and Stream
+    Handlers to analyze and summarize stream data, which is then made available for collection via REST API.
 
     Taps and Collection Policies may be created by passing the appropriate YAML configuration file to
     --config, and/or by enabling the admin REST API with --admin-api and using the appropriate endpoints.
@@ -47,40 +48,45 @@ static const char USAGE[] =
     For more documentation, see https://pktvisor.dev
 
     Base Options:
-      -d                    Daemonize; fork and continue running in the background [default: false]
-      -h --help             Show this screen
-      -v                    Verbose log output
-      --no-track            Don't send lightweight, anonymous usage metrics
-      --version             Show version
+      -d                          Daemonize; fork and continue running in the background [default: false]
+      -h --help                   Show this screen
+      -v                          Verbose log output
+      --no-track                  Don't send lightweight, anonymous usage metrics
+      --version                   Show version
     Web Server Options:
-      -l HOST               Run web server on the given host or IP [default: localhost]
-      -p PORT               Run web server on the given port [default: 10853]
-      --tls                 Enable TLS on the web server
-      --tls-cert FILE       Use given TLS cert. Required if --tls is enabled.
-      --tls-key FILE        Use given TLS private key. Required if --tls is enabled.
-      --admin-api           Enable admin REST API giving complete control plane functionality [default: false]
-                            When not specified, the exposed API is read-only access to module status and metrics.
-                            When specified, write access is enabled for all modules.
+      -l HOST                     Run web server on the given host or IP [default: localhost]
+      -p PORT                     Run web server on the given port [default: 10853]
+      --tls                       Enable TLS on the web server
+      --tls-cert FILE             Use given TLS cert. Required if --tls is enabled.
+      --tls-key FILE              Use given TLS private key. Required if --tls is enabled.
+      --admin-api                 Enable admin REST API giving complete control plane functionality [default: false]
+                                  When not specified, the exposed API is read-only access to module status and metrics.
+                                  When specified, write access is enabled for all modules.
     Geo Options:
-      --geo-city FILE       GeoLite2 City database to use for IP to Geo mapping
-      --geo-asn FILE        GeoLite2 ASN database to use for IP to ASN mapping
+      --geo-city FILE             GeoLite2 City database to use for IP to Geo mapping
+      --geo-asn FILE              GeoLite2 ASN database to use for IP to ASN mapping
     Configuration:
-      --config FILE         Use specified YAML configuration to configure options, Taps, and Collection Policies
-                            Please see https://pktvisor.dev for more information
+      --config FILE               Use specified YAML configuration to configure options, Taps, and Collection Policies
+                                  Please see https://pktvisor.dev for more information
+    Modules:
+      --module-list               List all modules which have been loaded (builtin and dynamic)
+      --module-load FILE          Load the specified dynamic module
+      --module-dir DIR            Set module search path
     Logging Options:
-      --log-file FILE       Log to the given output file name
-      --syslog              Log to syslog
+      --log-file FILE             Log to the given output file name
+      --syslog                    Log to syslog
     Prometheus Options:
-      --prometheus          Ignored, Prometheus output always enabled (left for backwards compatibility)
-      --prom-instance ID    Optionally set the 'instance' label to given ID
+      --prometheus                Ignored, Prometheus output always enabled (left for backwards compatibility)
+      --prom-instance ID          Optionally set the 'instance' label to given ID
     Handler Module Defaults:
-      --max-deep-sample N   Never deep sample more than N% of streams (an int between 0 and 100) [default: 100]
-      --periods P           Hold this many 60 second time periods of history in memory [default: 5]
+      --max-deep-sample N         Never deep sample more than N% of streams (an int between 0 and 100) [default: 100]
+      --periods P                 Hold this many 60 second time periods of history in memory [default: 5]
     pcap Input Module Options: (applicable to default policy when IFACE is specified only)
-      -b BPF                Filter packets using the given tcpdump compatible filter expression. Example: "port 53"
-      -H HOSTSPEC           Specify subnets (comma separated) to consider HOST, in CIDR form. In live capture this /may/ be detected automatically
-                            from capture device but /must/ be specified for pcaps. Example: "10.0.1.0/24,10.0.2.1/32,2001:db8::/64"
-                            Specifying this for live capture will append to any automatic detection.
+      -b BPF                      Filter packets using the given tcpdump compatible filter expression. Example: "port 53"
+      -H HOSTSPEC                 Specify subnets (comma separated) to consider HOST, in CIDR form. In live capture this
+                                  /may/ be detected automatically from capture device but /must/ be specified for pcaps.
+                                  Example: "10.0.1.0/24,10.0.2.1/32,2001:db8::/64"
+                                  Specifying this for live capture will append to any automatic detection.
 )";
 
 namespace {
@@ -228,6 +234,47 @@ int main(int argc, char *argv[])
         logger->set_level(spdlog::level::debug);
     }
 
+    // modules
+    CoreRegistry registry;
+    if (args["--module-dir"]) {
+        registry.input_plugin_registry()->setPluginDirectory(args["--module-dir"].asString());
+        registry.handler_plugin_registry()->setPluginDirectory(args["--module-dir"].asString());
+    }
+    if (args["--module-load"]) {
+        auto meta = registry.input_plugin_registry()->metadata(args["--module-load"].asString());
+        if (!meta) {
+            logger->error("failed to load plugin: {}", args["--module-load"].asString());
+            exit(EXIT_FAILURE);
+        }
+        if (!meta->data().hasValue("type") || (meta->data().value("type") != "handler" && meta->data().value("type") != "input")) {
+            logger->error("plugin configuration metadata did not specify a valid plugin type", args["--module-load"].asString());
+            exit(EXIT_FAILURE);
+        }
+        if (meta->data().value("type") == "input") {
+            auto result = registry.input_plugin_registry()->load(args["--module-load"].asString());
+            if (result != Corrade::PluginManager::LoadState::Loaded) {
+                logger->error("failed to load input plugin: {}", result);
+                exit(EXIT_FAILURE);
+            }
+        }
+        else if (meta->data().value("type") == "handler") {
+            auto result = registry.handler_plugin_registry()->load(args["--module-load"].asString());
+            if (result != Corrade::PluginManager::LoadState::Loaded) {
+                logger->error("failed to load input handler plugin: {}", result);
+                exit(EXIT_FAILURE);
+            }
+        }
+    }
+    if (args["--module-list"].asBool()) {
+        for (auto &p : registry.input_plugin_registry()->pluginList()) {
+            logger->info("input: {}", p);
+        }
+        for (auto &p : registry.handler_plugin_registry()->pluginList()) {
+            logger->info("handler: {}", p);
+        }
+        exit(EXIT_SUCCESS);
+    }
+
     logger->info("{} starting up", VISOR_VERSION);
 
     // if we are demonized, change to root directory now that (potentially) logs are open
@@ -256,7 +303,7 @@ int main(int argc, char *argv[])
 
     std::unique_ptr<CoreServer> svr;
     try {
-        svr = std::make_unique<CoreServer>(logger, http_config, prom_config);
+        svr = std::make_unique<CoreServer>(&registry, logger, http_config, prom_config);
     } catch (const std::exception &e) {
         logger->error(e.what());
         logger->info("exit with failure");

integration_tests/CMakeLists.txt

@@ -2,20 +2,30 @@
 set(FIXTURE_DIR ${CMAKE_SOURCE_DIR}/src/tests/fixtures)
 set(HOST_VAR "-H 127.0.0.1/32")
 set(PCAP_BINARY ${CMAKE_BINARY_DIR}/bin/pktvisor-pcap)
+set(PKTD_BINARY ${CMAKE_BINARY_DIR}/bin/pktvisord)
+set(VISOR_DYNMODULE_DIR ${CMAKE_BINARY_DIR}/lib/)
 set(INT_SH ${CMAKE_SOURCE_DIR}/integration_tests/integration.sh)
 set(WORKING_DIR ${CMAKE_SOURCE_DIR}/integration_tests)
 
-macro(visor_int_test name)
+macro(visor_pcap_int_test name)
     add_test(NAME ${name}
             WORKING_DIRECTORY ${WORKING_DIR}
             COMMAND ${INT_SH} ${PCAP_BINARY} -- ${HOST_VAR} --geo-city ${FIXTURE_DIR}/GeoIP2-City-Test.mmdb --geo-asn ${FIXTURE_DIR}/GeoIP2-ISP-Test.mmdb ${FIXTURE_DIR}/${name}.pcap)
 endmacro()
 
-visor_int_test(dns_ipv4_udp)
-visor_int_test(dns_ipv4_tcp)
-visor_int_test(dns_ipv6_udp)
-visor_int_test(dns_ipv6_tcp)
-visor_int_test(dhcp-flow)
+macro(visor_dyn_mod_int_test name)
+    add_test(NAME ${name}
+            WORKING_DIRECTORY ${WORKING_DIR}
+            COMMAND ${PKTD_BINARY} --module-dir ${VISOR_DYNMODULE_DIR} --module-load ${name} --module-list)
+endmacro()
+
+visor_pcap_int_test(dns_ipv4_udp)
+visor_pcap_int_test(dns_ipv4_tcp)
+visor_pcap_int_test(dns_ipv6_udp)
+visor_pcap_int_test(dns_ipv6_tcp)
+visor_pcap_int_test(dhcp-flow)
+
+visor_dyn_mod_int_test(VisorHandlerMock)
 
 # this allows local, non-public integration tests (for example, on private pcap data)
 #add_test(NAME external-tests

src/CoreRegistry.cpp

@@ -12,22 +12,33 @@
 
 namespace visor {
 
-CoreRegistry::CoreRegistry(HttpServer *svr)
-    : _svr(svr)
+CoreRegistry::CoreRegistry()
 {
 
     _logger = spdlog::get("visor");
     if (!_logger) {
         _logger = spdlog::stderr_color_mt("visor");
     }
 
+    // inputs
+    _input_manager = std::make_unique<InputStreamManager>();
+
+    // handlers
+    _handler_manager = std::make_unique<HandlerManager>();
+
+    // taps
+    _tap_manager = std::make_unique<TapManager>(this);
+
+    // policies policies
+    _policy_manager = std::make_unique<PolicyManager>(this);
+}
+
+void CoreRegistry::start(HttpServer *svr)
+{
     if (!svr) {
         _logger->warn("initializing modules with no HttpServer");
     }
 
-    // inputs
-    _input_manager = std::make_unique<InputStreamManager>();
-
     // initialize input plugins
     {
         auto alias_list = _input_registry.aliasList();
@@ -38,14 +49,11 @@ CoreRegistry::CoreRegistry(HttpServer *svr)
         for (auto &s : by_alias) {
             InputPluginPtr mod = _input_registry.instantiate(s);
             _logger->info("Load input stream plugin: {} {}", s, mod->pluginInterface());
-            mod->init_plugin(this, _svr);
+            mod->init_plugin(this, svr);
             _input_plugins.insert({s, std::move(mod)});
         }
     }
 
-    // handlers
-    _handler_manager = std::make_unique<HandlerManager>();
-
     // initialize handler plugins
     {
         auto alias_list = _handler_registry.aliasList();
@@ -56,15 +64,11 @@ CoreRegistry::CoreRegistry(HttpServer *svr)
         for (auto &s : by_alias) {
             HandlerPluginPtr mod = _handler_registry.instantiate(s);
             _logger->info("Load stream handler plugin: {} {}", s, mod->pluginInterface());
-            mod->init_plugin(this, _svr);
+            mod->init_plugin(this, svr);
             _handler_plugins.insert({s, std::move(mod)});
         }
     }
 
-    // taps
-    _tap_manager = std::make_unique<TapManager>(this);
-    // policies policies
-    _policy_manager = std::make_unique<PolicyManager>(this);
 }
 
 void CoreRegistry::stop()

src/CoreRegistry.h

@@ -38,7 +38,7 @@ class CoreRegistry
     HandlerPluginMap _handler_plugins;
 
     // these hold instances of active visor::AbstractModule derived modules (the main event processors) which are created from the plugins above
-    // any number can exist per plugin type can exist at a time, each with their own life cycle
+    // any number can exist per plugin type at a time, each with their own life cycle
     std::unique_ptr<InputStreamManager> _input_manager;
     std::unique_ptr<HandlerManager> _handler_manager;
 
@@ -47,12 +47,12 @@ class CoreRegistry
     std::unique_ptr<PolicyManager> _policy_manager;
 
     std::shared_ptr<spdlog::logger> _logger;
-    HttpServer *_svr;
 
 public:
-    CoreRegistry(HttpServer *svr);
+    CoreRegistry();
     ~CoreRegistry();
 
+    void start(HttpServer *svr);
     void stop();
 
     // yaml based configuration
@@ -84,10 +84,22 @@ class CoreRegistry
     {
         return _policy_manager.get();
     }
+    [[nodiscard]] const HandlerPluginRegistry *handler_plugin_registry() const
+    {
+        return &_handler_registry;
+    }
     [[nodiscard]] const InputPluginRegistry *input_plugin_registry() const
     {
         return &_input_registry;
     }
+    [[nodiscard]] HandlerPluginRegistry *handler_plugin_registry()
+    {
+        return &_handler_registry;
+    }
+    [[nodiscard]] InputPluginRegistry *input_plugin_registry()
+    {
+        return &_input_registry;
+    }
 
     [[nodiscard]] InputPluginMap &input_plugins()
     {