zeek-dev November 2017

zeek-dev@lists.zeek.org

6 discussions

[Bro-Dev] Feedback on configuration framework implementation

by Johanna Amann

Hello everyone, the branch topic/johanna/config contains an implementation of the configuration framework as it was discussed in an earlier thread on this list. GitHub link: https://github.com/bro/bro/compare/topic/johanna/config The implementation is basically what we discussed in the earlier thread with some additional components like a reader for configuration values and a script-level framework. It would be great if people could take a look at all of this and see if this makes sense, or if they see any problems with the implementation as it is at the moment. In the rest of the mail I wil go into a bit more detail and describe the different parts of this change. Note that the rest of this email will be very similar to the git commit message which also describes this change :) The configuration framework consists of three mostly distinct parts: * option variables * the config reader * the script level framework option variable =============== The option keyword allows variables to be specified as run-tine options. Such variables cannot be changed using normal assignments. Instead, they can be changed using Option::set. It is possible to "subscribe" to options and be notified when an option value changes. Change handlers can also change values before they are applied; this gives them the opportunity to reject changes. Priorities can be specified if there are several handlers for one option. Example script: option testbool: bool = T; function option_changed(ID: string, new_value: bool): bool { print fmt("Value of %s changed from %s to %s", ID, testbool, new_value); return new_value; } event bro_init() { print "Old value", testbool; Option::set_change_handler("testbool", option_changed); Option::set("testbool", F); print "New value", testbool; } config reader ============= The config reader provides a way to read configuration files back into Bro. Most importantly it automatically converts values to the correct types. This is important because it is at least inconvenient (and sometimes near impossible) to perform the necessary type conversions in Bro scripts themselves. This is especially true for sets/vectors. Configuration generally look like this: [option name][tab/spaces][new variable value] so, for example: testaddr 2607:f8b0:4005:801::200e testinterval 60 testtime 1507321987 test_set a b c d erdbeerschnitzel The reader uses the option name to look up the type that variable has in the Bro core and automatically converts the value to the correct type. Example script use: type Idx: record { option_name: string; }; type Val: record { option_val: string; }; global currconfig: table[string] of string = table(); event InputConfig::new_value(name: string, source: string, id: string, value: any) { print id, value; } event bro_init() { Input::add_table([$reader=Input::READER_CONFIG, $source="../configfile", $name="configuration", $idx=Idx, $val=Val, $destination=currconfig, $want_record=F]); } Script-level config framework ============================= The script-level framework ties these two features together and makes them a bit more convenient to use. Configuration files can simply be specified by placing them into Config::config_files. The framework also creates a config.log that shows all value changes that took place. Usage example: redef Config::config_files += {configfile}; export { option testbool : bool = F; } The file is now monitored for changes; when a change occurs the respective option values are automatically updated and the value change is written to config.log. Other changes ============= Internally, this commit also performs a range of changes to the Input manager; it marks a lot of functions as const and introduces a new ValueToVal method (which could in theory replace the already existing one - it is a bit more powerful). This also changes SerialTypes to have a subtype for Values, just as Fields already have it; I think it was mostly an oversight that this was not introduced from the beginning. This should not necessitate any code changes for people already using SerialTypes. Johanna

2 years, 11 months

[Bro-Dev] Log::set_buf Ineffective Before First Write

by Logan Lembke

Hello, I am currently writing a Bro IDS logging plugin for logging to MongoDB. We have implemented both buffered and unbuffered writes and rely on WriterBackend::DoSetBuf to be called in order to switch between the approaches. Currently, we use a bro script which attaches our plugin to the Conn log and calls Log::set_buf in order to configure the buffering behavior. However, DoSetBuf never gets called on our plugin. In Manager.cc, Manager::SetBuf loops over the list of writers registered with a given stream and calls the SetBuf method on each of the WriterFrontends. Unfortunately, this list of registered writers is empty before the first write, as writers are initialized as they are needed in the Manager::Write method. Effectively, this prevents configuring buffering behavior before the first write occurs. I'm new to the Bro code base, but I believe a fix could be made by storing the buffering behavior on the stream and checking this behavior on writer initialization. Here is the bro script I am currently using, The source code for the plugin is at https://github.com/ocmdev/bro-mongodb/tree/optionalBuffer. Does this look like a valid problem? Logan

3 years

[Bro-Dev] How to analyse the source code?

by Neil

I want do any modify about Bro. Howevery I don't know how to analyse the source code? Can you offer some document about the source code? Whice tools did you used to editor and compile the code? Thank you very much!

3 years

Re: [Bro-Dev] [Bro-Commits] [git/bro] topic/actor-system: First-pass broker-enabled Cluster scripting API + misc. (07ad06b)

by Robin Sommer

This is coming together quite nicely. Not sure if it's stable yet, but I'll just go ahead with some feedback I noticed looking over the new cluster API: - One thing I can't quite tell is if this is still aiming to maintain compatibility with the old communication system, like by keeping the proxies and also the *_events patterns. Looking at setup-connections, it seems so. I'd say just go ahead and remove all legacy pieces. Maintain two schemes in parallel is cumbersome, and I think it's fine to just force everything over to Broker. - Is the idea for the "*_topic" constants that one just picks the apppropiate one when sending events? Like if I want to publish something to all workers, I'd publish to Cluster::worker_topic? I think that's fine, though I'm wondering if we could compress the API there somehow so that Cluster doesn't need to export all those constants indvidiually. One idea would be a function that returns a topic based on node type? - I like the Pools! How about moving Pool with its functions out of the main.bro, just for clarity. - Looks like the hello/bye events are broadcasted by all nodes. Is that on purpose, or should that be limited to just one, like just the master sending them out? Or does it not matter and this provides for more redundancy? - create_store() vs "stores": Is the idea that I'd normally use create_store() and that populates the table, but I could also redef it myself instead of using create_store() to create more custom entries? If so, maybe make that a bit more explicit in the comments that there're two ways to configure that table. Robin On Fri, Oct 27, 2017 at 12:44 -0500, Jonathan Siwek wrote: > Repository : ssh://git@bro-ids.icir.org/bro > On branch : topic/actor-system > Link : https://github.com/bro/bro/commit/07ad06b083d16f9cf1c86041cf7287335a74ebbb > > >--------------------------------------------------------------- > > commit 07ad06b083d16f9cf1c86041cf7287335a74ebbb > Author: Jon Siwek <jsiwek(a)illinois.edu> > Date: Fri Oct 27 12:44:54 2017 -0500 > > First-pass broker-enabled Cluster scripting API + misc. > > - Remove Broker::Options, Broker::configure(). This was only > half implemented (e.g. the "routable" flag wasn't used), and using > a function to set these options was awkward: the only way to > override defaults was via calling configure() in a bro_init with > *lower* priority such that the last call "wins". Also doesn't > really make sense for it to be a function since the underlying > broker library can't adapt to changes in these configuration > values dynamically at runtime, so instead there's just now > two options you can redef: "Broker::forward_messages" and > "Broker::log_topic". > > - Add Broker::node_id() to get a unique identifier for the Bro instance's > broker endpoint. This is used by the Cluster API to map node name > (e.g. "manager") to broker endpoint so that one can track which nodes > are still alive. > > - Fix how broker-based communication interacts with --pseudo-realtime > and reading pcaps: bro now terminates at end of reading pcap when > broker is active (this should now be equivalent to how RemoteSerializer > worked). > > - New broker-enabled Cluster framework API > - Still uses Cluster::nodes as the means of setting up cluster network > - See Cluster::stores, Cluster::StoreInfo, and Cluster::create_store > for how broker data stores are integrated into cluster operation > > - Update several unit tests to new Cluster API. Failing tests at > the moment are mostly from scripts/frameworks that haven't been > ported to the new Cluster API. > > > >--------------------------------------------------------------- > > 07ad06b083d16f9cf1c86041cf7287335a74ebbb > aux/broker | 2 +- > scripts/base/frameworks/broker/main.bro | 47 ++-- > scripts/base/frameworks/broker/store.bro | 14 +- > scripts/base/frameworks/cluster/__load__.bro | 7 - > scripts/base/frameworks/cluster/main.bro | 279 ++++++++++++++++++++- > .../base/frameworks/cluster/setup-connections.bro | 150 +++++++++++ > scripts/base/frameworks/control/main.bro | 1 + > src/broker/Manager.cc | 20 +- > src/broker/Manager.h | 15 +- > src/broker/comm.bif | 21 +- > src/iosource/PktSrc.cc | 20 +- > testing/btest/Baseline/plugins.hooks/output | 18 +- > .../manager-1..stdout | 4 + > .../manager-1.reporter.log | 11 +- > testing/btest/broker/remote_log_types.bro | 2 +- > .../base/frameworks/cluster/start-it-up-logger.bro | 47 +++- > .../base/frameworks/cluster/start-it-up.bro | 48 +++- > .../frameworks/control/configuration_update.bro | 2 +- > .../scripts/base/frameworks/control/id_value.bro | 2 +- > .../scripts/base/frameworks/control/shutdown.bro | 2 +- > .../logging/field-extension-cluster-error.bro | 7 +- > .../frameworks/logging/field-extension-cluster.bro | 7 +- > 22 files changed, 601 insertions(+), 125 deletions(-) > > diff --git a/aux/broker b/aux/broker > index 76375d0..e1d637c 160000 > --- a/aux/broker > +++ b/aux/broker > @@ -1 +1 @@ > -Subproject commit 76375d07f5bf1ffc9711e064644bf865eda7a828 > +Subproject commit e1d637c816955a451079b419f438307960109346 > diff --git a/scripts/base/frameworks/broker/main.bro b/scripts/base/frameworks/broker/main.bro > index 47533a4..3ac25ed 100644 > --- a/scripts/base/frameworks/broker/main.bro > +++ b/scripts/base/frameworks/broker/main.bro > @@ -20,6 +20,11 @@ export { > ## use already. > const default_listen_retry = 30sec &redef; > > + ## Default address on which to listen. > + ## > + ## .. bro:see:: Broker::listen > + const default_listen_address = "" &redef; > + > ## Default interval to retry connecting to a peer if it cannot be made to work > ## initially, or if it ever becomes disconnected. > const default_connect_retry = 30sec &redef; > @@ -55,14 +60,12 @@ export { > ## all peers. > const ssl_keyfile = "" &redef; > > - ## The available configuration options when enabling Broker. > - type Options: record { > - ## Whether this Broker instance relays messages not destined to itself. > - ## By default, routing is disabled. > - routable: bool &default = F; > - ## The topic prefix where to publish logs. > - log_topic: string &default = "bro/logs/"; > - }; > + ## Forward all received messages to subscribing peers. > + const forward_messages = F &redef; > + > + ## The topic prefix where logs will be published. The log's stream id > + ## is appended when writing to a particular stream. > + const log_topic = "bro/logs/" &redef; > > type ErrorCode: enum { > ## The unspecified default error code. > @@ -153,13 +156,6 @@ export { > val: Broker::Data; > }; > > - ## Configures the local endpoint. > - ## > - ## options: Configures the local Broker endpoint behavior. > - ## > - ## Returns: true if configuration was successfully performed.. > - global configure: function(options: Options &default = Options()): bool; > - > ## Listen for remote connections. > ## > ## a: an address string on which to accept connections, e.g. > @@ -174,7 +170,8 @@ export { > ## Returns: the bound port or 0/? on failure. > ## > ## .. bro:see:: Broker::status > - global listen: function(a: string &default = "", p: port &default=default_port, > + global listen: function(a: string &default = default_listen_address, > + p: port &default = default_port, > retry: interval &default = default_listen_retry): port; > ## Initiate a remote connection. > ## > @@ -213,6 +210,9 @@ export { > ## Returns: a list of all peer connections. > global peers: function(): vector of PeerInfo; > > + ## Returns: a unique identifier for the local broker endpoint. > + global node_id: function(): string; > + > ## Publishes an event at a given topic. > ## > ## topic: a topic associated with the event message. > @@ -278,16 +278,6 @@ export { > > module Broker; > > -event bro_init() &priority=-10 > - { > - configure(); # Configure with defaults. > - } > - > -function configure(options: Options &default = Options()): bool > - { > - return __configure(options); > - } > - > event retry_listen(a: string, p: port, retry: interval) > { > listen(a, p, retry); > @@ -318,6 +308,11 @@ function peers(): vector of PeerInfo > return __peers(); > } > > +function node_id(): string > + { > + return __node_id(); > + } > + > function publish(topic: string, ev: Event): bool > { > return __publish(topic, ev); > diff --git a/scripts/base/frameworks/broker/store.bro b/scripts/base/frameworks/broker/store.bro > index ed735e0..6b22f5d 100644 > --- a/scripts/base/frameworks/broker/store.bro > +++ b/scripts/base/frameworks/broker/store.bro > @@ -61,7 +61,7 @@ export { > options: BackendOptions &default = BackendOptions()): opaque of Broker::Store; > > ## Create a clone of a master data store which may live with a remote peer. > - ## A clone automatically synchronizes to the master by automatically > + ## A clone automatically synchronizes to the master by > ## receiving modifications and applying them locally. Direct modifications > ## are not possible, they must be sent through the master store, which then > ## automatically broadcasts the changes out to clones. But queries may be > @@ -70,18 +70,6 @@ export { > ## > ## name: the unique name which identifies the master data store. > ## > - ## b: the storage backend to use. > - ## > - ## options: tunes how some storage backends operate. > - ## > - ## resync: the interval at which to re-attempt synchronizing with the master > - ## store should the connection be lost. If the clone has not yet > - ## synchronized for the first time, updates and queries queue up > - ## until the synchronization completes. After, if the connection > - ## to the master store is lost, queries continue to use the clone's > - ## version, but updates will be lost until the master is once again > - ## available. > - ## > ## Returns: a handle to the data store. > global create_clone: function(name: string): opaque of Broker::Store; > > diff --git a/scripts/base/frameworks/cluster/__load__.bro b/scripts/base/frameworks/cluster/__load__.bro > index 1717b83..4f193c0 100644 > --- a/scripts/base/frameworks/cluster/__load__.bro > +++ b/scripts/base/frameworks/cluster/__load__.bro > @@ -19,13 +19,6 @@ redef peer_description = Cluster::node; > > @load ./setup-connections > > -# Don't load the listening script until we're a bit more sure that the > -# cluster framework is actually being enabled. > -@load frameworks/communication/listen > - > -## Set the port that this node is supposed to listen on. > -redef Communication::listen_port = Cluster::nodes[Cluster::node]$p; > - > @if ( Cluster::local_node_type() == Cluster::MANAGER ) > @load ./nodes/manager > # If no logger is defined, then the manager receives logs. > diff --git a/scripts/base/frameworks/cluster/main.bro b/scripts/base/frameworks/cluster/main.bro > index 261f3f1..c94ed28 100644 > --- a/scripts/base/frameworks/cluster/main.bro > +++ b/scripts/base/frameworks/cluster/main.bro > @@ -7,10 +7,96 @@ > ##! ``@load base/frameworks/cluster``. > > @load base/frameworks/control > +@load base/frameworks/broker > > module Cluster; > > export { > + ## Whether the cluster framework uses broker to perform remote communication. > + const use_broker = T &redef; > + > + ## The topic name used for exchanging general messages that are relevant to > + ## any node in a cluster. Used with broker-enabled cluster communication. > + const broadcast_topic = "bro/cluster/broadcast" &redef; > + > + ## The topic name used for exchanging messages that are relevant to > + ## logger nodes in a cluster. Used with broker-enabled cluster communication. > + const logger_topic = "bro/cluster/logger" &redef; > + > + ## The topic name used for exchanging messages that are relevant to > + ## manager nodes in a cluster. Used with broker-enabled cluster communication. > + const manager_topic = "bro/cluster/manager" &redef; > + > + ## The topic name used for exchanging messages that are relevant to > + ## proxy nodes in a cluster. Used with broker-enabled cluster communication. > + const proxy_topic = "bro/cluster/proxy" &redef; > + > + ## The topic name used for exchanging messages that are relevant to > + ## worker nodes in a cluster. Used with broker-enabled cluster communication. > + const worker_topic = "bro/cluster/worker" &redef; > + > + ## The topic name used for exchanging messages that are relevant to > + ## time machine nodes in a cluster. Used with broker-enabled cluster communication. > + const time_machine_topic = "bro/cluster/time_machine" &redef; > + > + ## The topic prefix used for exchanging messages that are relevant to > + ## a named node in a cluster. Used with broker-enabled cluster communication. > + const node_topic_prefix = "bro/cluster/node/" &redef; > + > + ## Name of the node on which master data stores will be created if no other > + ## has already been specified by the user in :bro:see:`Cluster::stores`. > + const default_master_node = "manager" &redef; > + > + ## The type of data store backend that will be used for all data stores if > + ## no other has already been specified by the user in :bro:see:`Cluster::stores`. > + const default_backend = Broker::MEMORY &redef; > + > + ## The type of persistent data store backend that will be used for all data > + ## stores if no other has already been specified by the user in > + ## :bro:see:`Cluster::stores`. This will be used when script authors call > + ## :bro:see:`Cluster::create_store` with the *persistent* argument set true. > + const default_persistent_backend = Broker::SQLITE &redef; > + > + ## Setting a default dir will, for persistent backends that have not > + ## been given an explicit file path via :bro:see:`Cluster::stores`, > + ## automatically create a path within this dir that is based on the name of > + ## the data store. > + const default_store_dir = "" &redef; > + > + ## Information regarding a cluster-enabled data store. > + type StoreInfo: record { > + ## The name of the data store. > + name: string &optional; > + ## The store handle. > + store: opaque of Broker::Store &optional; > + ## The name of the cluster node on which the master version of the data > + ## store resides. > + master_node: string &default=default_master_node; > + ## Whether the data store is the master version or a clone. > + master: bool &default=F; > + ## The type of backend used for storing data. > + backend: Broker::BackendType &default=default_backend; > + ## Parameters used for configuring the backend. > + options: Broker::BackendOptions &default=Broker::BackendOptions(); > + }; > + > + ## A table of cluster-enabled data stores that have been created, indexed > + ## by their name. To customize a particular data store, you may redef this, > + ## defining the :bro:see:`StoreInfo` to associate with the store's name. > + global stores: table[string] of StoreInfo &default=StoreInfo() &redef; > + > + ## Sets up a cluster-enabled data store. They will also still properly > + ## function for uses that are not operating a cluster. > + ## > + ## name: the name of the data store to create. > + ## > + ## persistent: whether the data store must be persistent. > + ## > + ## Returns: the store's information. For master stores, the store will be > + ## ready to use immediately. For clones, the store field will not > + ## be set until the node containing the master store has connected. > + global create_store: function(name: string, persistent: bool &default=F): StoreInfo; > + > ## The cluster logging stream identifier. > redef enum Log::ID += { LOG }; > > @@ -18,6 +104,8 @@ export { > type Info: record { > ## The time at which a cluster message was generated. > ts: time; > + ## The name of the node that is creating the log record. > + node: string; > ## A message indicating information about the cluster's operation. > message: string; > } &log; > @@ -92,8 +180,7 @@ export { > ## If the *ip* field is a non-global IPv6 address, this field > ## can specify a particular :rfc:`4007` ``zone_id``. > zone_id: string &default=""; > - ## The port to which this local node can connect when > - ## establishing communication. > + ## The port that this node will listen on for peer connections. > p: port; > ## Identifier for the interface a worker is sniffing. > interface: string &optional; > @@ -108,6 +195,8 @@ export { > workers: set[string] &optional; > ## Name of a time machine node with which this node connects. > time_machine: string &optional; > + ## A unique identifier assigned to the node by the broker framework. > + id: string &optional; > }; > > ## This function can be called at any time to determine if the cluster > @@ -134,6 +223,8 @@ export { > ## named cluster-layout.bro somewhere in the BROPATH. It will be > ## automatically loaded if the CLUSTER_NODE environment variable is set. > ## Note that BroControl handles all of this automatically. > + ## The table is typically indexed by node names/labels (e.g. "manager" > + ## or "worker-1"). > const nodes: table[string] of Node = {} &redef; > > ## Indicates whether or not the manager will act as the logger and receive > @@ -148,6 +239,15 @@ export { > > ## Interval for retrying failed connections between cluster nodes. > const retry_interval = 1min &redef; > + > + ## When using broker-enabled cluster framework, nodes use this event to > + ## exchange their user-defined name along with a string that uniquely > + ## identifies it for the duration of its lifetime (this string may change if > + ## the node dies and has to reconnect later). > + global hello: event(name: string, id: string); > + > + ## Write a message to the cluster logging stream. > + global log: function(msg: string); > } > > function is_enabled(): bool > @@ -163,13 +263,112 @@ function local_node_type(): NodeType > event remote_connection_handshake_done(p: event_peer) &priority=5 > { > if ( p$descr in nodes && nodes[p$descr]$node_type == WORKER ) > - ++worker_count; > + { > + if ( use_broker ) > + Reporter::error(fmt("broker-enabled cluster using old comms: '%s' ", node)); > + else > + ++worker_count; > + } > } > > event remote_connection_closed(p: event_peer) &priority=5 > { > if ( p$descr in nodes && nodes[p$descr]$node_type == WORKER ) > - --worker_count; > + { > + if ( use_broker ) > + Reporter::error(fmt("broker-enabled cluster using old comms: '%s' ", node)); > + else > + --worker_count; > + } > + } > + > +event Cluster::hello(name: string, id: string) &priority=10 > + { > + if ( name !in nodes ) > + { > + Reporter::error(fmt("Got Cluster::hello msg from unexpected node: %s", name)); > + return; > + } > + > + local n = nodes[name]; > + > + if ( n?$id && n$id != id ) > + Reporter::error(fmt("Got Cluster::hello msg from duplicate node: %s", name)); > + > + n$id = id; > + Cluster::log(fmt("got hello from %s (%s)", name, id)); > + > + if ( n$node_type == WORKER ) > + ++worker_count; > + > + for ( store_name in stores ) > + { > + local info = stores[store_name]; > + > + if ( info?$store ) > + next; > + > + if ( info$master ) > + next; > + > + if ( info$master_node == name ) > + { > + info$store = Broker::create_clone(info$name); > + Cluster::log(fmt("created clone store: %s", info$name)); > + } > + } > + } > + > +event Broker::peer_added(endpoint: Broker::EndpointInfo, msg: string) &priority=10 > + { > + if ( ! use_broker ) > + return; > + > + if ( ! Cluster::is_enabled() ) > + return; > + > + local e = Broker::make_event(Cluster::hello, node, Broker::node_id()); > + Broker::publish(Cluster::broadcast_topic, e); > + } > + > +event Broker::peer_lost(endpoint: Broker::EndpointInfo, msg: string) &priority=10 > + { > + if ( ! use_broker ) > + return; > + > + for ( node_name in nodes ) > + { > + local n = nodes[node_name]; > + > + if ( n?$id && n$id == endpoint$id ) > + { > + Cluster::log(fmt("node down: %s", node_name)); > + delete n$id; > + > + if ( n$node_type == WORKER ) > + --worker_count; > + > + for ( store_name in stores ) > + { > + local info = stores[store_name]; > + > + if ( ! info?$store ) > + next; > + > + if ( info$master ) > + next; > + > + if ( info$master_node == node_name ) > + { > + Broker::close(info$store); > + delete info$store; > + Cluster::log(fmt("clone store closed: %s", info$name)); > + } > + } > + > + break; > + } > + } > } > > event bro_init() &priority=5 > @@ -183,3 +382,75 @@ event bro_init() &priority=5 > > Log::create_stream(Cluster::LOG, [$columns=Info, $path="cluster"]); > } > + > +function create_store(name: string, persistent: bool &default=F): Cluster::StoreInfo > + { > + local info = stores[name]; > + info$name = name; > + > + if ( Cluster::default_store_dir != "" ) > + { > + local default_options = Broker::BackendOptions(); > + local path = Cluster::default_store_dir + "/" + name; > + > + if ( info$options$sqlite$path == default_options$sqlite$path ) > + info$options$sqlite$path = path + ".sqlite"; > + > + if ( info$options$rocksdb$path == default_options$rocksdb$path ) > + info$options$rocksdb$path = path + ".rocksdb"; > + } > + > + if ( persistent ) > + { > + switch ( info$backend ) { > + case Broker::MEMORY: > + info$backend = Cluster::default_persistent_backend; > + break; > + case Broker::SQLITE: > + fallthrough; > + case Broker::ROCKSDB: > + # no-op: user already asked for a specific persistent backend. > + break; > + default: > + Reporter::error(fmt("unhandled data store type: %s", info$backend)); > + break; > + } > + } > + > + if ( ! Cluster::is_enabled() ) > + { > + if ( info?$store ) > + { > + Reporter::warning(fmt("duplicate cluster store creation for %s", name)); > + return info; > + } > + > + info$store = Broker::create_master(name, info$backend, info$options); > + info$master = T; > + stores[name] = info; > + Cluster::log(fmt("created master store: %s", name)); > + return info; > + } > + > + if ( info$master_node !in Cluster::nodes ) > + Reporter::fatal(fmt("master node '%s' for cluster store '%s' does not exist", > + info$master_node, name)); > + > + if ( Cluster::node == info$master_node ) > + { > + info$store = Broker::create_master(name, info$backend, info$options); > + info$master = T; > + stores[name] = info; > + return info; > + } > + > + info$master = F; > + stores[name] = info; > + Cluster::log(fmt("pending clone store creation: %s", name)); > + return info; > + } > + > +function log(msg: string) > + { > + Log::write(Cluster::LOG, [$ts = network_time(), $node = node, $message = msg]); > + } > diff --git a/scripts/base/frameworks/cluster/setup-connections.bro b/scripts/base/frameworks/cluster/setup-connections.bro > index 971a55d..2e775da 100644 > --- a/scripts/base/frameworks/cluster/setup-connections.bro > +++ b/scripts/base/frameworks/cluster/setup-connections.bro > @@ -3,13 +3,163 @@ > > @load ./main > @load base/frameworks/communication > +@load base/frameworks/broker > > @if ( Cluster::node in Cluster::nodes ) > > module Cluster; > > +type NamedNode: record { > + name: string; > + node: Node; > +}; > + > +function nodes_with_type(node_type: NodeType): vector of NamedNode > + { > + local rval: vector of NamedNode = vector(); > + > + for ( name in Cluster::nodes ) > + { > + local n = Cluster::nodes[name]; > + > + if ( n$node_type != node_type ) > + next; > + > + rval[|rval|] = NamedNode($name=name, $node=n); > + } > + > + return rval; > + } > + > +function connect_peer(node_type: NodeType, node_name: string): bool > + { > + local nn = nodes_with_type(node_type); > + > + for ( i in nn ) > + { > + local n = nn[i]; > + > + if ( n$name != node_name ) > + next; > + > + Cluster::log(fmt("initiate peering with %s:%s, retry=%s", > + n$node$ip, n$node$p, Cluster::retry_interval)); > + return Broker::peer(cat(n$node$ip), n$node$p, Cluster::retry_interval); > + } > + > + return F; > + } > + > event bro_init() &priority=9 > { > + if ( ! use_broker ) > + return; > + > + local self = nodes[node]; > + > + switch ( self$node_type ) { > + case NONE: > + return; > + case CONTROL: > + break; > + case LOGGER: > + Broker::subscribe(Cluster::logger_topic); > + Broker::subscribe(Broker::log_topic); > + break; > + case MANAGER: > + Broker::subscribe(Cluster::manager_topic); > + > + if ( Cluster::manager_is_logger ) > + Broker::subscribe(Broker::log_topic); > + > + break; > + case PROXY: > + Broker::subscribe(Cluster::proxy_topic); > + break; > + case WORKER: > + Broker::subscribe(Cluster::worker_topic); > + break; > + case TIME_MACHINE: > + Broker::subscribe(Cluster::time_machine_topic); > + break; > + default: > + Reporter::error(fmt("Unhandled cluster node type: %s", self$node_type)); > + return; > + } > + > + Broker::subscribe(Cluster::broadcast_topic); > + Broker::subscribe(Cluster::node_topic_prefix + node); > + > + Broker::listen(Broker::default_listen_address, > + self$p, > + Broker::default_listen_retry); > + > + Cluster::log(fmt("listening on %s:%s", Broker::default_listen_address, self$p)); > + > + switch ( self$node_type ) { > + case MANAGER: > + if ( self?$logger ) > + connect_peer(LOGGER, self$logger); > + > + if ( self?$time_machine ) > + connect_peer(TIME_MACHINE, self$time_machine); > + > + break; > + case PROXY: > + if ( self?$logger ) > + connect_peer(LOGGER, self$logger); > + > + if ( self?$manager ) > + connect_peer(MANAGER, self$manager); > + > + local proxies = nodes_with_type(PROXY); > + > + for ( i in proxies ) > + { > + local proxy = proxies[i]; > + > + if ( proxy$node?$proxy ) > + Broker::peer(cat(proxy$node$ip), proxy$node$p, Cluster::retry_interval); > + } > + > + break; > + case WORKER: > + if ( self?$logger ) > + connect_peer(LOGGER, self$logger); > + > + if ( self?$manager ) > + connect_peer(MANAGER, self$manager); > + > + if ( self?$proxy ) > + connect_peer(PROXY, self$proxy); > + > + if ( self?$time_machine ) > + connect_peer(TIME_MACHINE, self$time_machine); > + > + break; > + } > + } > + > +event bro_init() &priority=-10 > + { > + if ( use_broker ) > + return; > + > + local lp = Cluster::nodes[Cluster::node]$p; > + enable_communication(); > + listen(Communication::listen_interface, > + lp, > + Communication::listen_ssl, > + Communication::listen_ipv6, > + Communication::listen_ipv6_zone_id, > + Communication::listen_retry); > + } > + > +event bro_init() &priority=9 > + { > + if ( use_broker ) > + return; > + > local me = nodes[node]; > > for ( i in Cluster::nodes ) > diff --git a/scripts/base/frameworks/control/main.bro b/scripts/base/frameworks/control/main.bro > index 5c68c47..e3b58ef 100644 > --- a/scripts/base/frameworks/control/main.bro > +++ b/scripts/base/frameworks/control/main.bro > @@ -5,6 +5,7 @@ > module Control; > > export { > + ## Whether the control framework uses broker to perform remote communication. > const use_broker = T &redef; > > ## The address of the host that will be controlled. > diff --git a/src/broker/Manager.cc b/src/broker/Manager.cc > index e9d6280..b2cbe87 100644 > --- a/src/broker/Manager.cc > +++ b/src/broker/Manager.cc > @@ -113,7 +113,6 @@ Manager::BrokerState::BrokerState(broker::broker_options options) > > Manager::Manager() > { > - routable = false; > bound_port = 0; > > next_timestamp = 1; > @@ -128,6 +127,7 @@ void Manager::InitPostScript() > { > DBG_LOG(DBG_BROKER, "Initializing"); > > + log_topic = get_option("Broker::log_topic")->AsString()->CheckString(); > log_id_type = internal_type("Log::ID")->AsEnumType(); > writer_id_type = internal_type("Log::Writer")->AsEnumType(); > > @@ -144,6 +144,7 @@ void Manager::InitPostScript() > > broker::broker_options options; > options.disable_ssl = get_option("Broker::disable_ssl")->AsBool(); > + options.forward = get_option("Broker::forward_messages")->AsBool(); > > bstate = std::make_shared<BrokerState>(options); > } > @@ -176,18 +177,6 @@ bool Manager::Active() > return bound_port > 0 || bstate->endpoint.peers().size(); > } > > -bool Manager::Configure(bool arg_routable, std::string arg_log_topic) > - { > - DBG_LOG(DBG_BROKER, "Configuring endpoint: routable=%s log_topic=%s", > - (routable ? "yes" : "no"), arg_log_topic.c_str()); > - > - routable = arg_routable; > - log_topic = arg_log_topic; > - > - // TODO: process routable flag > - return true; > - } > - > uint16_t Manager::Listen(const string& addr, uint16_t port) > { > bound_port = bstate->endpoint.listen(addr, port); > @@ -233,6 +222,11 @@ std::vector<broker::peer_info> Manager::Peers() const > return bstate->endpoint.peers(); > } > > +std::string Manager::NodeID() const > + { > + return to_string(bstate->endpoint.node_id()); > + } > + > bool Manager::PublishEvent(string topic, std::string name, broker::vector args) > { > if ( ! bstate->endpoint.peers().size() ) > diff --git a/src/broker/Manager.h b/src/broker/Manager.h > index a430c57..5af23e4 100644 > --- a/src/broker/Manager.h > +++ b/src/broker/Manager.h > @@ -74,15 +74,6 @@ public: > bool Active(); > > /** > - * Configure the local Broker endpoint. > - * @param routable Whether the context of this endpoint routes messages not > - * @param log_topic The topic prefix for logs we this endpoint published. > - * destined to itself. By default endpoints do not route. > - * @return true if configuration was successful. > - */ > - bool Configure(bool routable = false, std::string log_topic=""); > - > - /** > * Listen for remote connections. > * @param port the TCP port to listen on. > * @param addr an address string on which to accept connections, e.g. > @@ -115,6 +106,11 @@ public: > std::vector<broker::peer_info> Peers() const; > > /** > + * @return a unique identifier for this broker endpoint. > + */ > + std::string NodeID() const; > + > + /** > * Send an event to any interested peers. > * @param topic a topic string associated with the message. > * Peers advertise interest by registering a subscription to some prefix > @@ -296,7 +292,6 @@ private: > broker::endpoint& Endpoint() > { assert(bstate); return bstate->endpoint; } > > - bool routable; > std::string log_topic; > uint16_t bound_port; > > diff --git a/src/broker/comm.bif b/src/broker/comm.bif > index 411e3d4..c7c94d4 100644 > --- a/src/broker/comm.bif > +++ b/src/broker/comm.bif > @@ -7,8 +7,6 @@ > > module Broker; > > -type Broker::Options: record; > - > ## Generated when something changes in the Broker sub-system. > event Broker::status%(endpoint: EndpointInfo, msg: string%); > > @@ -51,20 +49,6 @@ enum PeerStatus %{ > RECONNECTING, > %} > > -function Broker::__configure%(options: Broker::Options%): bool > - %{ > - auto routable = false; > - auto log_topic = ""; > - > - if ( auto routable_val = options->AsRecordVal()->Lookup(0) ) > - routable = routable_val->AsBool(); > - > - if ( auto log_topic_val = options->AsRecordVal()->Lookup(1) ) > - log_topic = log_topic_val->AsString()->CheckString(); > - > - return new Val(broker_mgr->Configure(routable, log_topic), TYPE_BOOL); > - %} > - > function Broker::__listen%(a: string, p: port%): port > %{ > if ( ! p->IsTCP() ) > @@ -140,3 +124,8 @@ function Broker::__peers%(%): PeerInfos > > return rval; > %} > + > +function Broker::__node_id%(%): string > + %{ > + return new StringVal(broker_mgr->NodeID()); > + %} > diff --git a/src/iosource/PktSrc.cc b/src/iosource/PktSrc.cc > index a9362a0..343801a 100644 > --- a/src/iosource/PktSrc.cc > +++ b/src/iosource/PktSrc.cc > @@ -10,6 +10,8 @@ > #include "Hash.h" > #include "Net.h" > #include "Sessions.h" > +#include "broker/Manager.h" > +#include "iosource/Manager.h" > > #include "pcap/pcap.bif.h" > > @@ -304,13 +306,19 @@ bool PktSrc::ExtractNextPacketInternal() > return 1; > } > > - if ( pseudo_realtime && using_communication && ! IsOpen() ) > + if ( pseudo_realtime && ! IsOpen() ) > { > - // Source has gone dry, we're done. > - if ( remote_trace_sync_interval ) > - remote_serializer->SendFinalSyncPoint(); > - else > - remote_serializer->Terminate(); > + if ( using_communication ) > + { > + // Source has gone dry, we're done. > + if ( remote_trace_sync_interval ) > + remote_serializer->SendFinalSyncPoint(); > + else > + remote_serializer->Terminate(); > + } > + > + if ( broker_mgr->Active() ) > + iosource_mgr->Terminate(); > } > > SetIdle(true); > diff --git a/testing/btest/Baseline/plugins.hooks/output b/testing/btest/Baseline/plugins.hooks/output > index f7c61c0..5eefc64 100644 > --- a/testing/btest/Baseline/plugins.hooks/output > +++ b/testing/btest/Baseline/plugins.hooks/output > @@ -148,8 +148,6 @@ > 0.000000 MetaHookPost CallFunction(Analyzer::register_for_ports, <frame>, (Analyzer::ANALYZER_SYSLOG, {514/udp})) -> <no result> > 0.000000 MetaHookPost CallFunction(Analyzer::register_for_ports, <frame>, (Analyzer::ANALYZER_TEREDO, {3544/udp})) -> <no result> > 0.000000 MetaHookPost CallFunction(Analyzer::register_for_ports, <frame>, (Analyzer::ANALYZER_XMPP, {5222<...>/tcp})) -> <no result> > -0.000000 MetaHookPost CallFunction(Broker::__configure, <frame>, ([routable=F, log_topic=bro<...>/])) -> <no result> > -0.000000 MetaHookPost CallFunction(Broker::configure, <frame>, ([routable=F, log_topic=bro<...>/])) -> <no result> > 0.000000 MetaHookPost CallFunction(Cluster::is_enabled, <frame>, ()) -> <no result> > 0.000000 MetaHookPost CallFunction(Cluster::is_enabled, <null>, ()) -> <no result> > 0.000000 MetaHookPost CallFunction(Files::register_analyzer_add_callback, <frame>, (Files::ANALYZER_EXTRACT, FileExtract::on_add{ if (!FileExtract::args?$extract_filename) FileExtract::args$extract_filename = cat(extract-, FileExtract::f$last_active, -, FileExtract::f$source, -, FileExtract::f$id)FileExtract::f$info$extracted = FileExtract::args$extract_filenameFileExtract::args$extract_filename = build_path_compressed(FileExtract::prefix, FileExtract::args$extract_filename)FileExtract::f$info$extracted_cutoff = Fmkdir(FileExtract::prefix)})) -> <no result> > @@ -251,7 +249,7 @@ > 0.000000 MetaHookPost CallFunction(Log::__create_stream, <frame>, (Weird::LOG, [columns=<no value description>, ev=Weird::log_weird, path=weird])) -> <no result> > 0.000000 MetaHookPost CallFunction(Log::__create_stream, <frame>, (X509::LOG, [columns=<no value description>, ev=X509::log_x509, path=x509])) -> <no result> > 0.000000 MetaHookPost CallFunction(Log::__create_stream, <frame>, (mysql::LOG, [columns=<no value description>, ev=MySQL::log_mysql, path=mysql])) -> <no result> > -0.000000 MetaHookPost CallFunction(Log::__write, <frame>, (PacketFilter::LOG, [ts=1502745368.796663, node=bro, filter=ip or not ip, init=T, success=T])) -> <no result> > +0.000000 MetaHookPost CallFunction(Log::__write, <frame>, (PacketFilter::LOG, [ts=1509124227.694371, node=bro, filter=ip or not ip, init=T, success=T])) -> <no result> > 0.000000 MetaHookPost CallFunction(Log::add_default_filter, <frame>, (Broker::LOG)) -> <no result> > 0.000000 MetaHookPost CallFunction(Log::add_default_filter, <frame>, (Cluster::LOG)) -> <no result> > 0.000000 MetaHookPost CallFunction(Log::add_default_filter, <frame>, (Communication::LOG)) -> <no result> > @@ -384,7 +382,7 @@ > 0.000000 MetaHookPost CallFunction(Log::create_stream, <frame>, (Weird::LOG, [columns=<no value description>, ev=Weird::log_weird, path=weird])) -> <no result> > 0.000000 MetaHookPost CallFunction(Log::create_stream, <frame>, (X509::LOG, [columns=<no value description>, ev=X509::log_x509, path=x509])) -> <no result> > 0.000000 MetaHookPost CallFunction(Log::create_stream, <frame>, (mysql::LOG, [columns=<no value description>, ev=MySQL::log_mysql, path=mysql])) -> <no result> > -0.000000 MetaHookPost CallFunction(Log::write, <frame>, (PacketFilter::LOG, [ts=1502745368.796663, node=bro, filter=ip or not ip, init=T, success=T])) -> <no result> > +0.000000 MetaHookPost CallFunction(Log::write, <frame>, (PacketFilter::LOG, [ts=1509124227.694371, node=bro, filter=ip or not ip, init=T, success=T])) -> <no result> > 0.000000 MetaHookPost CallFunction(NetControl::check_plugins, <frame>, ()) -> <no result> > 0.000000 MetaHookPost CallFunction(NetControl::init, <null>, ()) -> <no result> > 0.000000 MetaHookPost CallFunction(Notice::want_pp, <frame>, ()) -> <no result> > @@ -872,8 +870,6 @@ > 0.000000 MetaHookPre CallFunction(Analyzer::register_for_ports, <frame>, (Analyzer::ANALYZER_SYSLOG, {514/udp})) > 0.000000 MetaHookPre CallFunction(Analyzer::register_for_ports, <frame>, (Analyzer::ANALYZER_TEREDO, {3544/udp})) > 0.000000 MetaHookPre CallFunction(Analyzer::register_for_ports, <frame>, (Analyzer::ANALYZER_XMPP, {5222<...>/tcp})) > -0.000000 MetaHookPre CallFunction(Broker::__configure, <frame>, ([routable=F, log_topic=bro<...>/])) > -0.000000 MetaHookPre CallFunction(Broker::configure, <frame>, ([routable=F, log_topic=bro<...>/])) > 0.000000 MetaHookPre CallFunction(Cluster::is_enabled, <frame>, ()) > 0.000000 MetaHookPre CallFunction(Cluster::is_enabled, <null>, ()) > 0.000000 MetaHookPre CallFunction(Files::register_analyzer_add_callback, <frame>, (Files::ANALYZER_EXTRACT, FileExtract::on_add{ if (!FileExtract::args?$extract_filename) FileExtract::args$extract_filename = cat(extract-, FileExtract::f$last_active, -, FileExtract::f$source, -, FileExtract::f$id)FileExtract::f$info$extracted = FileExtract::args$extract_filenameFileExtract::args$extract_filename = build_path_compressed(FileExtract::prefix, FileExtract::args$extract_filename)FileExtract::f$info$extracted_cutoff = Fmkdir(FileExtract::prefix)})) > @@ -975,7 +971,7 @@ > 0.000000 MetaHookPre CallFunction(Log::__create_stream, <frame>, (Weird::LOG, [columns=<no value description>, ev=Weird::log_weird, path=weird])) > 0.000000 MetaHookPre CallFunction(Log::__create_stream, <frame>, (X509::LOG, [columns=<no value description>, ev=X509::log_x509, path=x509])) > 0.000000 MetaHookPre CallFunction(Log::__create_stream, <frame>, (mysql::LOG, [columns=<no value description>, ev=MySQL::log_mysql, path=mysql])) > -0.000000 MetaHookPre CallFunction(Log::__write, <frame>, (PacketFilter::LOG, [ts=1502745368.796663, node=bro, filter=ip or not ip, init=T, success=T])) > +0.000000 MetaHookPre CallFunction(Log::__write, <frame>, (PacketFilter::LOG, [ts=1509124227.694371, node=bro, filter=ip or not ip, init=T, success=T])) > 0.000000 MetaHookPre CallFunction(Log::add_default_filter, <frame>, (Broker::LOG)) > 0.000000 MetaHookPre CallFunction(Log::add_default_filter, <frame>, (Cluster::LOG)) > 0.000000 MetaHookPre CallFunction(Log::add_default_filter, <frame>, (Communication::LOG)) > @@ -1108,7 +1104,7 @@ > 0.000000 MetaHookPre CallFunction(Log::create_stream, <frame>, (Weird::LOG, [columns=<no value description>, ev=Weird::log_weird, path=weird])) > 0.000000 MetaHookPre CallFunction(Log::create_stream, <frame>, (X509::LOG, [columns=<no value description>, ev=X509::log_x509, path=x509])) > 0.000000 MetaHookPre CallFunction(Log::create_stream, <frame>, (mysql::LOG, [columns=<no value description>, ev=MySQL::log_mysql, path=mysql])) > -0.000000 MetaHookPre CallFunction(Log::write, <frame>, (PacketFilter::LOG, [ts=1502745368.796663, node=bro, filter=ip or not ip, init=T, success=T])) > +0.000000 MetaHookPre CallFunction(Log::write, <frame>, (PacketFilter::LOG, [ts=1509124227.694371, node=bro, filter=ip or not ip, init=T, success=T])) > 0.000000 MetaHookPre CallFunction(NetControl::check_plugins, <frame>, ()) > 0.000000 MetaHookPre CallFunction(NetControl::init, <null>, ()) > 0.000000 MetaHookPre CallFunction(Notice::want_pp, <frame>, ()) > @@ -1596,8 +1592,6 @@ > 0.000000 | HookCallFunction Analyzer::register_for_ports(Analyzer::ANALYZER_SYSLOG, {514/udp}) > 0.000000 | HookCallFunction Analyzer::register_for_ports(Analyzer::ANALYZER_TEREDO, {3544/udp}) > 0.000000 | HookCallFunction Analyzer::register_for_ports(Analyzer::ANALYZER_XMPP, {5222<...>/tcp}) > -0.000000 | HookCallFunction Broker::__configure([routable=F, log_topic=bro<...>/]) > -0.000000 | HookCallFunction Broker::configure([routable=F, log_topic=bro<...>/]) > 0.000000 | HookCallFunction Cluster::is_enabled() > 0.000000 | HookCallFunction Files::register_analyzer_add_callback(Files::ANALYZER_EXTRACT, FileExtract::on_add{ if (!FileExtract::args?$extract_filename) FileExtract::args$extract_filename = cat(extract-, FileExtract::f$last_active, -, FileExtract::f$source, -, FileExtract::f$id)FileExtract::f$info$extracted = FileExtract::args$extract_filenameFileExtract::args$extract_filename = build_path_compressed(FileExtract::prefix, FileExtract::args$extract_filename)FileExtract::f$info$extracted_cutoff = Fmkdir(FileExtract::prefix)}) > 0.000000 | HookCallFunction Files::register_for_mime_type(Files::ANALYZER_PE, application/x-dosexec) > @@ -1698,7 +1692,7 @@ > 0.000000 | HookCallFunction Log::__create_stream(Weird::LOG, [columns=<no value description>, ev=Weird::log_weird, path=weird]) > 0.000000 | HookCallFunction Log::__create_stream(X509::LOG, [columns=<no value description>, ev=X509::log_x509, path=x509]) > 0.000000 | HookCallFunction Log::__create_stream(mysql::LOG, [columns=<no value description>, ev=MySQL::log_mysql, path=mysql]) > -0.000000 | HookCallFunction Log::__write(PacketFilter::LOG, [ts=1502745368.796663, node=bro, filter=ip or not ip, init=T, success=T]) > +0.000000 | HookCallFunction Log::__write(PacketFilter::LOG, [ts=1509124227.694371, node=bro, filter=ip or not ip, init=T, success=T]) > 0.000000 | HookCallFunction Log::add_default_filter(Broker::LOG) > 0.000000 | HookCallFunction Log::add_default_filter(Cluster::LOG) > 0.000000 | HookCallFunction Log::add_default_filter(Communication::LOG) > @@ -1831,7 +1825,7 @@ > 0.000000 | HookCallFunction Log::create_stream(Weird::LOG, [columns=<no value description>, ev=Weird::log_weird, path=weird]) > 0.000000 | HookCallFunction Log::create_stream(X509::LOG, [columns=<no value description>, ev=X509::log_x509, path=x509]) > 0.000000 | HookCallFunction Log::create_stream(mysql::LOG, [columns=<no value description>, ev=MySQL::log_mysql, path=mysql]) > -0.000000 | HookCallFunction Log::write(PacketFilter::LOG, [ts=1502745368.796663, node=bro, filter=ip or not ip, init=T, success=T]) > +0.000000 | HookCallFunction Log::write(PacketFilter::LOG, [ts=1509124227.694371, node=bro, filter=ip or not ip, init=T, success=T]) > 0.000000 | HookCallFunction NetControl::check_plugins() > 0.000000 | HookCallFunction NetControl::init() > 0.000000 | HookCallFunction Notice::want_pp() > diff --git a/testing/btest/Baseline/scripts.base.frameworks.cluster.start-it-up/manager-1..stdout b/testing/btest/Baseline/scripts.base.frameworks.cluster.start-it-up/manager-1..stdout > index 7c8eb5e..5b10602 100644 > --- a/testing/btest/Baseline/scripts.base.frameworks.cluster.start-it-up/manager-1..stdout > +++ b/testing/btest/Baseline/scripts.base.frameworks.cluster.start-it-up/manager-1..stdout > @@ -1,4 +1,8 @@ > Connected to a peer > Connected to a peer > Connected to a peer > +Got fully_connected event > +Got fully_connected event > Connected to a peer > +Got fully_connected event > +Got fully_connected event > diff --git a/testing/btest/Baseline/scripts.base.frameworks.logging.field-extension-cluster-error/manager-1.reporter.log b/testing/btest/Baseline/scripts.base.frameworks.logging.field-extension-cluster-error/manager-1.reporter.log > index b7d8c11..e7972c2 100644 > --- a/testing/btest/Baseline/scripts.base.frameworks.logging.field-extension-cluster-error/manager-1.reporter.log > +++ b/testing/btest/Baseline/scripts.base.frameworks.logging.field-extension-cluster-error/manager-1.reporter.log > @@ -3,11 +3,10 @@ > #empty_field (empty) > #unset_field - > #path reporter > -#open 2016-09-22-23-31-34 > +#open 2017-10-26-19-18-59 > #fields _write_ts _stream _system_name ts level message location > #types time string string time enum string string > -1474587094.261799 reporter manager-1 0.000000 Reporter::WARNING WriterFrontend communication/Log::WRITER_ASCII expected 11 fields in write, got 8. Skipping line. (empty) > -1474587094.261799 reporter manager-1 0.000000 Reporter::WARNING WriterFrontend communication/Log::WRITER_ASCII expected 11 fields in write, got 8. Skipping line. (empty) > -1474587094.261799 reporter manager-1 0.000000 Reporter::WARNING WriterFrontend communication/Log::WRITER_ASCII expected 11 fields in write, got 8. Skipping line. (empty) > -1474587099.984660 reporter manager-1 0.000000 Reporter::INFO received termination signal (empty) > -#close 2016-09-22-23-31-40 > +1509045539.693078 reporter manager-1 0.000000 Reporter::WARNING Write using filter 'default' on path 'broker' changed to use new path 'broker-2' to avoid conflict with filter '' (empty) > +1509045539.699623 reporter manager-1 0.000000 Reporter::WARNING WriterFrontend cluster/Log::WRITER_ASCII expected 6 fields in write, got 3. Skipping line. (empty) > +1509045547.196521 reporter manager-1 0.000000 Reporter::INFO received termination signal (empty) > +#close 2017-10-26-19-19-07 > diff --git a/testing/btest/broker/remote_log_types.bro b/testing/btest/broker/remote_log_types.bro > index aeaf1b9..56d63eb 100644 > --- a/testing/btest/broker/remote_log_types.bro > +++ b/testing/btest/broker/remote_log_types.bro > @@ -1,4 +1,4 @@ > - @TEST-SERIALIZE: brokercomm > +# @TEST-SERIALIZE: brokercomm > > # @TEST-EXEC: btest-bg-run recv "bro -b ../recv.bro >recv.out" > # @TEST-EXEC: btest-bg-run send "bro -b ../send.bro >send.out" > diff --git a/testing/btest/scripts/base/frameworks/cluster/start-it-up-logger.bro b/testing/btest/scripts/base/frameworks/cluster/start-it-up-logger.bro > index 97f3698..72251c0 100644 > --- a/testing/btest/scripts/base/frameworks/cluster/start-it-up-logger.bro > +++ b/testing/btest/scripts/base/frameworks/cluster/start-it-up-logger.bro > @@ -1,4 +1,4 @@ > -# @TEST-SERIALIZE: comm > +# @TEST-SERIALIZE: brokercomm > # > # @TEST-EXEC: btest-bg-run logger-1 CLUSTER_NODE=logger-1 BROPATH=$BROPATH:.. bro %INPUT > # @TEST-EXEC: sleep 1 > @@ -38,10 +38,16 @@ global fully_connected_nodes = 0; > event fully_connected() > { > ++fully_connected_nodes; > + > if ( Cluster::node == "logger-1" ) > { > if ( peer_count == 5 && fully_connected_nodes == 5 ) > - terminate_communication(); > + { > + if ( Cluster::use_broker ) > + terminate(); > + else > + terminate_communication(); > + } > } > } > > @@ -49,6 +55,43 @@ redef Cluster::worker2logger_events += /fully_connected/; > redef Cluster::proxy2logger_events += /fully_connected/; > redef Cluster::manager2logger_events += /fully_connected/; > > +event bro_init() > + { > + Broker::auto_publish(Cluster::logger_topic, fully_connected); > + } > + > +event Broker::peer_added(endpoint: Broker::EndpointInfo, msg: string) > + { > + print "Connected to a peer"; > + ++peer_count; > + > + if ( Cluster::node == "logger-1" ) > + { > + if ( peer_count == 5 && fully_connected_nodes == 5 ) > + { > + if ( Cluster::use_broker ) > + terminate(); > + else > + terminate_communication(); > + } > + } > + else if ( Cluster::node == "manager-1" ) > + { > + if ( peer_count == 5 ) > + event fully_connected(); > + } > + else > + { > + if ( peer_count == 3 ) > + event fully_connected(); > + } > + } > + > +event Broker::peer_lost(endpoint: Broker::EndpointInfo, msg: string) > + { > + terminate(); > + } > + > event remote_connection_handshake_done(p: event_peer) > { > print "Connected to a peer"; > diff --git a/testing/btest/scripts/base/frameworks/cluster/start-it-up.bro b/testing/btest/scripts/base/frameworks/cluster/start-it-up.bro > index acb9c36..b0fcc69 100644 > --- a/testing/btest/scripts/base/frameworks/cluster/start-it-up.bro > +++ b/testing/btest/scripts/base/frameworks/cluster/start-it-up.bro > @@ -1,4 +1,4 @@ > -# @TEST-SERIALIZE: comm > +# @TEST-SERIALIZE: brokercomm > # > # @TEST-EXEC: btest-bg-run manager-1 BROPATH=$BROPATH:.. CLUSTER_NODE=manager-1 bro %INPUT > # @TEST-EXEC: sleep 1 > @@ -8,7 +8,7 @@ > # @TEST-EXEC: btest-bg-run worker-1 BROPATH=$BROPATH:.. CLUSTER_NODE=worker-1 bro %INPUT > # @TEST-EXEC: btest-bg-run worker-2 BROPATH=$BROPATH:.. CLUSTER_NODE=worker-2 bro %INPUT > # @TEST-EXEC: btest-bg-wait 30 > -# @TEST-EXEC: btest-diff manager-1/.stdout > +# @TEST-EXEC: TEST_DIFF_CANONIFIER=$SCRIPTS/diff-sort btest-diff manager-1/.stdout > # @TEST-EXEC: btest-diff proxy-1/.stdout > # @TEST-EXEC: btest-diff proxy-2/.stdout > # @TEST-EXEC: btest-diff worker-1/.stdout > @@ -32,17 +32,59 @@ global fully_connected_nodes = 0; > > event fully_connected() > { > + if ( ! is_remote_event() ) > + return; > + > + print "Got fully_connected event"; > fully_connected_nodes = fully_connected_nodes + 1; > + > if ( Cluster::node == "manager-1" ) > { > if ( peer_count == 4 && fully_connected_nodes == 4 ) > - terminate_communication(); > + { > + if ( Cluster::use_broker ) > + terminate(); > + else > + terminate_communication(); > + } > } > } > > redef Cluster::worker2manager_events += /fully_connected/; > redef Cluster::proxy2manager_events += /fully_connected/; > > +event bro_init() > + { > + Broker::auto_publish(Cluster::manager_topic, fully_connected); > + } > + > +event Broker::peer_added(endpoint: Broker::EndpointInfo, msg: string) > + { > + print "Connected to a peer"; > + peer_count = peer_count + 1; > + > + if ( Cluster::node == "manager-1" ) > + { > + if ( peer_count == 4 && fully_connected_nodes == 4 ) > + { > + if ( Cluster::use_broker ) > + terminate(); > + else > + terminate_communication(); > + } > + } > + else > + { > + if ( peer_count == 2 ) > + event fully_connected(); > + } > + } > + > +event Broker::peer_lost(endpoint: Broker::EndpointInfo, msg: string) > + { > + terminate(); > + } > + > event remote_connection_handshake_done(p: event_peer) > { > print "Connected to a peer"; > diff --git a/testing/btest/scripts/base/frameworks/control/configuration_update.bro b/testing/btest/scripts/base/frameworks/control/configuration_update.bro > index 535a357..8d8fb0a 100644 > --- a/testing/btest/scripts/base/frameworks/control/configuration_update.bro > +++ b/testing/btest/scripts/base/frameworks/control/configuration_update.bro > @@ -1,4 +1,4 @@ > -# @TEST-SERIALIZE: comm > +# @TEST-SERIALIZE: brokercomm > # > # @TEST-EXEC: btest-bg-run controllee BROPATH=$BROPATH:.. bro %INPUT frameworks/control/controllee Communication::listen_port=65531/tcp Broker::default_port=65531/tcp > # @TEST-EXEC: sleep 5 > diff --git a/testing/btest/scripts/base/frameworks/control/id_value.bro b/testing/btest/scripts/base/frameworks/control/id_value.bro > index bee37a2..1b5e354 100644 > --- a/testing/btest/scripts/base/frameworks/control/id_value.bro > +++ b/testing/btest/scripts/base/frameworks/control/id_value.bro > @@ -1,4 +1,4 @@ > -# @TEST-SERIALIZE: comm > +# @TEST-SERIALIZE: brokercomm > # > # @TEST-EXEC: btest-bg-run controllee BROPATH=$BROPATH:.. bro %INPUT only-for-controllee frameworks/control/controllee Communication::listen_port=65532/tcp Broker::default_port=65532/tcp > # @TEST-EXEC: btest-bg-run controller BROPATH=$BROPATH:.. bro %INPUT frameworks/control/controller Control::host=127.0.0.1 Control::host_port=65532/tcp Control::cmd=id_value Control::arg=test_var > diff --git a/testing/btest/scripts/base/frameworks/control/shutdown.bro b/testing/btest/scripts/base/frameworks/control/shutdown.bro > index 9c0c104..2869c3f 100644 > --- a/testing/btest/scripts/base/frameworks/control/shutdown.bro > +++ b/testing/btest/scripts/base/frameworks/control/shutdown.bro > @@ -1,4 +1,4 @@ > -# @TEST-SERIALIZE: comm > +# @TEST-SERIALIZE: brokercomm > # > # @TEST-EXEC: btest-bg-run controllee BROPATH=$BROPATH:.. bro %INPUT frameworks/control/controllee Communication::listen_port=65530/tcp Broker::default_port=65530/tcp > # @TEST-EXEC: btest-bg-run controller BROPATH=$BROPATH:.. bro %INPUT frameworks/control/controller Control::host=127.0.0.1 Control::host_port=65530/tcp Control::cmd=shutdown > diff --git a/testing/btest/scripts/base/frameworks/logging/field-extension-cluster-error.bro b/testing/btest/scripts/base/frameworks/logging/field-extension-cluster-error.bro > index 6ac7a5e..c0ab10b 100644 > --- a/testing/btest/scripts/base/frameworks/logging/field-extension-cluster-error.bro > +++ b/testing/btest/scripts/base/frameworks/logging/field-extension-cluster-error.bro > @@ -1,4 +1,4 @@ > -# @TEST-SERIALIZE: comm > +# @TEST-SERIALIZE: brokercomm > # > # @TEST-EXEC: btest-bg-run manager-1 "cp ../cluster-layout.bro . && CLUSTER_NODE=manager-1 bro %INPUT" > # @TEST-EXEC: sleep 1 > @@ -43,6 +43,11 @@ event terminate_me() { > terminate(); > } > > +event Broker::peer_lost(endpoint: Broker::EndpointInfo, msg: string) > + { > + schedule 1sec { terminate_me() }; > + } > + > event remote_connection_closed(p: event_peer) { > schedule 1sec { terminate_me() }; > } > diff --git a/testing/btest/scripts/base/frameworks/logging/field-extension-cluster.bro b/testing/btest/scripts/base/frameworks/logging/field-extension-cluster.bro > index fb51251..6740743 100644 > --- a/testing/btest/scripts/base/frameworks/logging/field-extension-cluster.bro > +++ b/testing/btest/scripts/base/frameworks/logging/field-extension-cluster.bro > @@ -1,4 +1,4 @@ > -# @TEST-SERIALIZE: comm > +# @TEST-SERIALIZE: brokercomm > # > # @TEST-EXEC: btest-bg-run manager-1 "cp ../cluster-layout.bro . && CLUSTER_NODE=manager-1 bro %INPUT" > # @TEST-EXEC: sleep 1 > @@ -39,6 +39,11 @@ event terminate_me() { > terminate(); > } > > +event Broker::peer_lost(endpoint: Broker::EndpointInfo, msg: string) > + { > + schedule 1sec { terminate_me() }; > + } > + > event remote_connection_closed(p: event_peer) { > schedule 1sec { terminate_me() }; > } > > > > _______________________________________________ > bro-commits mailing list > bro-commits(a)bro.org > http://mailman.icsi.berkeley.edu/mailman/listinfo/bro-commits > -- Robin Sommer * ICSI/LBNL * robin(a)icir.org * www.icir.org/robin

3 years

[Bro-Dev] Scientific notation?

by Seth Hall

Right now, Bro will print scientific notation in JSON logs but we've always tended to avoid it in the standard Bro log format. What does everyone think about switching to allow scientific notation in the standard log format? Daniel recently did some exploration of various versions of awk and they all support scientific notation (I think that was part of my concern a long time ago). Thoughts? .Seth -- Seth Hall * Corelight, Inc * www.corelight.com

3 years

[Bro-Dev] File Analysis Inconsistencies

by Aaron Eppert

I crafted a custom file analysis plugin that attaches to specific MIME types via file_sniff and fires an appropriate event once processing has been completed. I had to jump through a few hoops to make a file analysis plugin, first, but those were cleared and everything runs and loads appropriately there (bro -NN verified.) My test regime is very straight forward, I have several PCAPs cooked up that contain simple HTTP file GETs (that extract otherwise properly and do not exhibit missing_bytes) and I am running them via `bro -C -r <>.pcap`. My issue comes with utter and complete inconsistency with execution - it is, effectively, a coin flip, with zero changes. When I have dumped the buffers being processed, as my file analysis plugin has a secondary verification to make sure the data passed is appropriate - which is confusing, as the mime type fires correct, which seems to indicate a bug somewhere in the data path - the correct execution, clearly has the proper data in it. The invalid executions, again changing nothing other than a subsequent execution, shows a buffer of what appears to be completely random data. I currently cannot supply the file analysis plugin for inspection, but would very much appreciate insight in how to find the root cause. It very much seems to be upstream. If I run the analysis portion of the plugin as a free standing executable outside of Bro against the data transferred via HTTP, everything works perfect and the structures are filled accordingly. I saw BIT-1832, and there could be similar root causes in there, but I have not had time to investigate otherwise. The issues I am raising, again, are command line replay via command line, not even “live” network traffic or tcpreplay over a NIC/dummy interface. Aaron

3 years

2020

2019

2018

2017

2016

2015

2014

2013

2012

2011

2010

zeek-dev November 2017