Disable communication with a Zeek instance if it's package version is too old.

CMakeLists.txt

@@ -72,12 +72,6 @@ else ()
     set(ZEEK_AGENT_VERSION_LONG "${ZEEK_AGENT_VERSION}")
 endif ()
 
-string(REGEX MATCH "([0-9]*)\.([0-9]*)\.([0-9]*).*" _ ${ZEEK_AGENT_VERSION})
-set(ZEEK_AGENT_VERSION_MAJOR "${CMAKE_MATCH_1}")
-set(ZEEK_AGENT_VERSION_MINOR "${CMAKE_MATCH_2}")
-set(ZEEK_AGENT_VERSION_PATCH "${CMAKE_MATCH_3}")
-math(EXPR ZEEK_AGENT_VERSION_NUMBER "${ZEEK_AGENT_VERSION_MAJOR} * 10000 + ${ZEEK_AGENT_VERSION_MINOR} * 100 + ${ZEEK_AGENT_VERSION_PATCH}")
-
 ### Platform-specific code.
 
 set(HAVE_POSIX $<bool:${UNIX}>)
@@ -117,7 +111,7 @@ endif ()
 message(
     "\n====================|  Zeek Agent Build Summary  |===================="
     "\n"
-    "\nVersion:               ${ZEEK_AGENT_VERSION_LONG} (${ZEEK_AGENT_VERSION_NUMBER})"
+    "\nVersion:               ${ZEEK_AGENT_VERSION_LONG}"
     "\n"
     "\nBuild type:            ${CMAKE_BUILD_TYPE}"
     "\nBuild directory:       ${PROJECT_BINARY_DIR}"

src/config.h.in

@@ -11,7 +11,6 @@
 namespace zeek::agent {
 inline const auto Version = "${ZEEK_AGENT_VERSION}";
 inline const auto VersionLong = "${ZEEK_AGENT_VERSION_LONG}";
-inline const auto VersionNumber = ${ZEEK_AGENT_VERSION_NUMBER}l;
 inline const auto InstallLibDir = "${CMAKE_INSTALL_LIBDIR}";
 inline const auto InstallPrefix = "${CMAKE_INSTALL_PREFIX}";
 }

src/core/configuration.cc

@@ -80,6 +80,7 @@ static void usage(const filesystem::path& name) {
 }
 
 void Options::debugDump() {
+    ZEEK_AGENT_DEBUG("configuration", "[option] version-number: {}", version_number);
     ZEEK_AGENT_DEBUG("configuration", "[option] mode: {}", to_string(mode));
     ZEEK_AGENT_DEBUG("configuration", "[option] agent-id: {}", agent_id);
     ZEEK_AGENT_DEBUG("configuration", "[option] instance-id: {}", instance_id);
@@ -139,6 +140,12 @@ struct Pimpl<Configuration>::Implementation {
 Options Configuration::Implementation::default_() {
     Options options;
 
+    auto version = parseVersion(Version);
+    if ( ! version )
+        throw InternalError("cannot parse our own version number");
+
+    options.version_number = *version;
+
     // Attempt to read our agent's ID from previously created cache file.
     auto uuid_path = (platform::dataDirectory() / "uuid").native();
     if ( filesystem::is_regular_file(uuid_path) ) {
@@ -154,7 +161,6 @@ Options Configuration::Implementation::default_() {
     if ( options.agent_id.empty() ) {
         // Generate a fresh UUID as our agent's ID.
         options.agent_id = format("H{}", randomUUID());
-        ;
 
         // Cache it.
         std::ofstream out(uuid_path, std::ios::out | std::ios::trunc);

src/core/configuration.h

@@ -94,6 +94,9 @@ struct Options {
     /** Logs a summary of the current settings to the debug log stream. */
     void debugDump();
 
+    /** Agent's numerical version number. This is set automatically and cannot changed externally. */
+    int64_t version_number;
+
     /** Mode of operation for the current process. */
     options::Mode mode = options::Mode::Standard;
 

src/io/zeek.cc

@@ -31,6 +31,10 @@
 #include <broker/topic.hh>
 #include <broker/zeek.hh>
 
+// Minimum version of the Zeek-side package that we require. If we see an agent
+// with an older version, we'll stop communicating with it.
+static const int64_t MininumZeekPackageVersion = 200020008;
+
 // Helpers for debugging logging that include additional state.
 #define ZEEK_INSTANCE_DEBUG(instance, ...)                                                                             \
     ZEEK_AGENT_DEBUG("zeek", "{}", format("[{}/{}] ", endpoint(), instance) + format(__VA_ARGS__))
@@ -115,6 +119,7 @@ class BrokerConnection {
         std::string version_string;            // Zeek version string, from instance's hello
         uint64_t version_number = 0;           // Zeek version number, from instance's hello
         std::string package_version = "<n/a>"; // Zeek agent package version, from instance's hello
+        bool disabled = false;                 // If true, we won't send/process any activity to/from this agent
 
         bool operator==(const ZeekInstance& other) const {
             // Ignore last seen, we're interested only in semantic changes.
@@ -336,31 +341,52 @@ void BrokerConnection::processEvent(const broker::data_message& msg) {
         return;
     }
 
+    if ( zeek_instance->second.disabled ) {
+        ZEEK_INSTANCE_DEBUG(zeek_instance_id, "ignoring event from disabled Zeek: {}{}", event.name(),
+                            broker::to_string(event.args()));
+        return;
+    }
+
     ZEEK_INSTANCE_DEBUG(zeek_instance_id, "got event: {}{}", event.name(), broker::to_string(event.args()));
 
     assert(zeek_instance != _zeek_instances.end());
 
     if ( event.name() == "ZeekAgentAPI::zeek_hello_v1" ) {
-        if ( args.size() >= 2 ) { // TODO: make two args mandatory eventually
-            try {
-                auto old_hello_record = zeek_instance->second;
-
-                auto hello_record = broker::get<broker::vector>(args[1]);
-                zeek_instance->second.version_string = broker::get<std::string>(hello_record[0]);
-                zeek_instance->second.version_number = broker::get<uint64_t>(hello_record[1]);
-
-                if ( auto pkg_version = broker::get<std::string>(hello_record[2]); ! pkg_version.empty() )
-                    zeek_instance->second.package_version = pkg_version;
-
-                if ( zeek_instance->second != old_hello_record ) {
-                    ZEEK_INSTANCE_DEBUG(zeek_instance_id, "Zeek version: {} ({}), package {}",
-                                        zeek_instance->second.version_string, zeek_instance->second.version_number,
-                                        zeek_instance->second.package_version);
+        try {
+            auto old_hello_record = zeek_instance->second;
+
+            auto hello_record = broker::get<broker::vector>(args[1]);
+            zeek_instance->second.version_string = broker::get<std::string>(hello_record[0]);
+            zeek_instance->second.version_number = broker::get<uint64_t>(hello_record[1]);
+
+            if ( auto pkg_version = broker::get<std::string>(hello_record[2]); ! pkg_version.empty() ) {
+                zeek_instance->second.package_version = pkg_version;
+
+                if ( auto pkg_version_number = parseVersion(pkg_version) ) {
+                    if ( *pkg_version_number < MininumZeekPackageVersion ) {
+                        const auto msg =
+                            format("Zeek package version too old, disabling communication (want {}, but have {})",
+                                   MininumZeekPackageVersion, *pkg_version_number);
+                        logger()->warn("[{}] {}", zeek_instance_id, msg);
+
+                        // We'll try to get the error message through still.
+                        transmitError(zeek_instance_id, msg, {}, {});
+                        zeek_instance->second.disabled = true;
+                        return;
+                    }
                 }
-            } catch ( const std::exception& e ) {
-                unexpectedEventArguments(zeek_instance_id, event);
-                return;
+                else
+                    ZEEK_INSTANCE_DEBUG(zeek_instance_id, "cannot parse Zeek package version number ({})", pkg_version);
+            }
+
+            if ( zeek_instance->second != old_hello_record ) {
+                ZEEK_INSTANCE_DEBUG(zeek_instance_id, "Zeek version: {} ({}), package {}",
+                                    zeek_instance->second.version_string, zeek_instance->second.version_number,
+                                    zeek_instance->second.package_version);
             }
+        } catch ( const std::exception& e ) {
+            unexpectedEventArguments(zeek_instance_id, event);
+            return;
         }
     }
 
@@ -612,6 +638,11 @@ void BrokerConnection::transmitResult(const std::string& zeek_id, const query::R
 void BrokerConnection::transmitError(const std::string& zeek_instance, const std::string& msg,
                                      const std::optional<std::string>& zeek_id,
                                      const std::optional<std::string>& cookie) {
+    if ( auto i = _zeek_instances.find(zeek_instance); i != _zeek_instances.end() && i->second.disabled ) {
+        ZEEK_INSTANCE_DEBUG(zeek_instance, "not sending error to disabled Zeek: {}", msg);
+        return;
+    }
+
     ZEEK_INSTANCE_DEBUG(zeek_instance, "sending error: {}", msg);
     transmitEvent("ZeekAgentAPI::agent_error_v1", {msg}, zeek_instance, zeek_id, cookie);
 }
@@ -624,6 +655,13 @@ void BrokerConnection::transmitEvent(std::string event_name, broker::vector args
     assert(! zeek_instance.has_value() || ! zeek_instance->empty());
     assert(! cookie.has_value() || ! cookie->empty());
 
+    if ( zeek_instance ) {
+        if ( auto i = _zeek_instances.find(*zeek_instance); i != _zeek_instances.end() && i->second.disabled ) {
+            ZEEK_INSTANCE_DEBUG(*zeek_instance, "not sending event {} to disabled Zeek", event_name);
+            return;
+        }
+    }
+
     broker::data change_data;
     if ( change ) {
         switch ( *change ) {

src/tables/zeek_agent/zeek_agent.darwin.mm

@@ -79,7 +79,7 @@ Value addresses() {
     Value addrs = addresses();
     Value platform = platform::name();
     Value os_name = std::string("macOS ") + std::string([version UTF8String]);
-    Value agent = VersionNumber;
+    Value agent = options().version_number;
     Value broker = broker::version::string();
     Value uptime = std::chrono::system_clock::now() - startupTime();
     Value tables =

src/tables/zeek_agent/zeek_agent.linux.cc

@@ -143,7 +143,7 @@ std::vector<std::vector<Value>> ZeekAgentLinux::snapshot(const std::vector<table
     Value addrs = addresses();
     Value platform = platform::name();
     Value os_name = distribution();
-    Value agent = VersionNumber;
+    Value agent = options().version_number;
     Value broker = broker::version::string();
     Value uptime = std::chrono::system_clock::now() - startupTime();
     Value tables =

src/tables/zeek_agent/zeek_agent.test.cc

@@ -11,5 +11,5 @@ TEST_CASE_FIXTURE(test::TableFixture, "zeek_agent" * doctest::test_suite("Tables
     useTable("zeek_agent");
 
     auto result = query("SELECT * from zeek_agent");
-    CHECK_EQ(result.get<int64_t>(0, "agent_version"), VersionNumber);
+    CHECK_EQ(result.get<int64_t>(0, "agent_version"), cfg.options().version_number);
 }

src/util/helpers.cc

@@ -125,6 +125,43 @@ std::string base62_encode(uint64_t i) {
     return x;
 }
 
+zeek::agent::Result<int64_t> parseVersion(std::string v) {
+    unsigned long major = 0;
+    unsigned long minor = 0;
+    unsigned long patch = 0;
+    unsigned long commit = 0;
+
+    if ( startsWith(v, "v") )
+        // ignore leading 'v'
+        v = v.substr(1);
+
+    try {
+        auto m = split(v, "-");
+        if ( m.empty() )
+            return result::Error("empty version string");
+
+        auto n = split(m[0], ".");
+        if ( n.size() != 3 )
+            return result::Error("not a valid version string");
+
+        major = std::stoul(n[0]);
+        minor = std::stoul(n[1]);
+        patch = std::stoul(n[2]);
+
+        if ( m.size() > 1 ) {
+            try {
+                commit = std::stoul(m[m.size() - 1]);
+            } catch ( ... ) {
+                // ignore errors
+            }
+        }
+
+        return static_cast<int64_t>(major * 100000000 + minor * 1000000 + patch * 10000 + commit);
+    } catch ( ... ) {
+        return result::Error("trouble parsing version string");
+    }
+}
+
 std::string randomUUID() {
     std::random_device rd;
     auto seed_data = std::array<int, std::mt19937::state_size>{};
@@ -300,4 +337,14 @@ TEST_SUITE("Helpers") {
             CHECK_EQ(split1("a b c"), str_pair("a", "b c"));
         }
     }
+
+    TEST_CASE("parseVersion") {
+        CHECK_EQ(parseVersion("2.0.4"), 200040000);
+        CHECK_EQ(parseVersion("2.0.4-123"), 200040123);
+        CHECK_EQ(parseVersion("2.0.4-rc1-123"), 200040123);
+        CHECK_EQ(parseVersion("2.0.4-rc1"), 200040000);
+        CHECK(! parseVersion(""));
+        CHECK(! parseVersion("x.x.x"));
+        CHECK(! parseVersion("x.x"));
+    }
 }

src/util/helpers.h

@@ -3,6 +3,7 @@
 #pragma once
 
 #include "util/fmt.h"
+#include "util/result.h"
 
 #include <algorithm>
 #include <chrono>
@@ -286,6 +287,9 @@ inline bool startsWith(const std::string& s, const std::string& prefix) { return
 /** Renders an integer in base62 ASCII. */
 std::string base62_encode(uint64_t i);
 
+/** Parsed a version string of the form `x.y.z-<N>` into a numerical number suitable for ordering. */
+zeek::agent::Result<int64_t> parseVersion(std::string v);
+
 /** Creates a new random UUID, encoded in base62 ASCII. */
 std::string randomUUID();