diff --git a/scripts/base/frameworks/broker/main.zeek b/scripts/base/frameworks/broker/main.zeek index 1213555dc5..1cd99b3771 100644 --- a/scripts/base/frameworks/broker/main.zeek +++ b/scripts/base/frameworks/broker/main.zeek @@ -264,6 +264,10 @@ export { type PeerInfo: record { peer: EndpointInfo; status: PeerStatus; + + ## Whether the local node created the peering, as opposed to a + ## remote establishing it by connecting to us. + is_outbound: bool; }; type PeerInfos: vector of PeerInfo; @@ -367,6 +371,16 @@ export { ## TODO: We do not have a function yet to terminate a connection. global unpeer: function(a: string, p: port): bool; + ## Whether the local node originally initiated the peering with the + ## given endpoint. + ## + ## a: the address used in previous successful call to :zeek:see:`Broker::peer`. + ## + ## p: the port used in previous successful call to :zeek:see:`Broker::peer`. + ## + ## Returns:: True if this node initiated the peering. + global is_outbound_peering: function(a: string, p: port): bool; + ## Get a list of all peer connections. ## ## Returns: a list of all peer connections. @@ -522,6 +536,11 @@ function unpeer(a: string, p: port): bool return __unpeer(a, p); } +function is_outbound_peering(a: string, p: port): bool + { + return __is_outbound_peering(a, p); + } + function peers(): vector of PeerInfo { return __peers(); diff --git a/src/broker/Manager.cc b/src/broker/Manager.cc index 3ea29b8b58..a77f050ede 100644 --- a/src/broker/Manager.cc +++ b/src/broker/Manager.cc @@ -233,6 +233,7 @@ public: LoggerQueuePtr loggerQueue; SeverityLevel logSeverity = SeverityLevel::critical; SeverityLevel stderrSeverity = SeverityLevel::critical; + std::unordered_set outbound_peerings; }; const broker::endpoint_info Manager::NoPeer{{}, {}}; @@ -582,6 +583,7 @@ void Manager::Peer(const string& addr, uint16_t port, double retry) { auto secs = broker::timeout::seconds(static_cast(retry)); bstate->endpoint.peer_nosync(addr, port, secs); + bstate->outbound_peerings.emplace(broker::network_info(addr, port)); auto counts_as_iosource = get_option("Broker::peer_counts_as_iosource")->AsBool(); @@ -613,6 +615,15 @@ void Manager::Unpeer(const string& addr, uint16_t port) { FlushLogBuffers(); bstate->endpoint.unpeer_nosync(addr, port); + bstate->outbound_peerings.erase(broker::network_info(addr, port)); +} + +bool Manager::IsOutboundPeering(const string& addr, uint16_t port) const { + return bstate->outbound_peerings.find(broker::network_info(addr, port)) != bstate->outbound_peerings.end(); +} + +bool Manager::IsOutboundPeering(const broker::network_info& ni) const { + return bstate->outbound_peerings.find(ni) != bstate->outbound_peerings.end(); } std::vector Manager::Peers() const { diff --git a/src/broker/Manager.h b/src/broker/Manager.h index d2dfbf2261..215e9efdf5 100644 --- a/src/broker/Manager.h +++ b/src/broker/Manager.h @@ -154,6 +154,21 @@ public: */ void Unpeer(const std::string& addr, uint16_t port); + /** + * Whether the local node originally initiated the peering with the + * given endpoint. + * @param addr the address used in zeek::Broker::Manager::Peer(). + * @param port the port used in zeek::Broker::Manager::Peer(). + */ + bool IsOutboundPeering(const std::string& addr, uint16_t port) const; + + /** + * Whether the local node originally initiated the peering with the + * given endpoint. + * @param ni the address and port used in zeek::Broker::Manager::Peer(). + */ + bool IsOutboundPeering(const broker::network_info& ni) const; + /** * @return a list of peer endpoints. */ diff --git a/src/broker/comm.bif b/src/broker/comm.bif index 81f1114a22..0fefce871d 100644 --- a/src/broker/comm.bif +++ b/src/broker/comm.bif @@ -203,6 +203,12 @@ function Broker::__unpeer%(a: string, p: port%): bool return zeek::val_mgr->True(); %} +function Broker::__is_outbound_peering%(a: string, p: port%): bool + %{ + zeek::Broker::Manager::ScriptScopeGuard ssg; + return zeek::val_mgr->Bool(broker_mgr->IsOutboundPeering(a->CheckString(), p->Port())); + %} + function Broker::__peers%(%): PeerInfos %{ zeek::Broker::Manager::ScriptScopeGuard ssg; @@ -237,6 +243,15 @@ function Broker::__peers%(%): PeerInfos peer_info->Assign(0, std::move(endpoint_info)); peer_info->Assign(1, zeek::BifType::Enum::Broker::PeerStatus->GetEnumVal(ps)); + // Broker has an existing concept of peer flags, see the broker::peer_info + // and broker::peer_flags structs. They currently aren't currently, but + // we can update the following logic once they are. + + if ( p.peer.network.has_value() ) + peer_info->Assign(2, zeek::val_mgr->Bool(broker_mgr->IsOutboundPeering(p.peer.network.value()))); + else + peer_info->Assign(2, zeek::val_mgr->False()); + rval->Assign(i, std::move(peer_info)); ++i; } diff --git a/testing/btest/Baseline/broker.peering-directionality/client.out b/testing/btest/Baseline/broker.peering-directionality/client.out new file mode 100644 index 0000000000..4dbe099979 --- /dev/null +++ b/testing/btest/Baseline/broker.peering-directionality/client.out @@ -0,0 +1,4 @@ +### BTest baseline data generated by btest-diff. Do not edit. Use "btest -U/-u" to update. Requires BTest >= 0.63. +peered, this is the outgoing peering: T +via Broker::peers(): T +after unpeering: F diff --git a/testing/btest/Baseline/broker.peering-directionality/server.out b/testing/btest/Baseline/broker.peering-directionality/server.out new file mode 100644 index 0000000000..e0845c9bf0 --- /dev/null +++ b/testing/btest/Baseline/broker.peering-directionality/server.out @@ -0,0 +1,3 @@ +### BTest baseline data generated by btest-diff. Do not edit. Use "btest -U/-u" to update. Requires BTest >= 0.63. +peered, this is the outgoing peering: F +via Broker::peers(): F diff --git a/testing/btest/Baseline/opt.ZAM-bif-tracking/output b/testing/btest/Baseline/opt.ZAM-bif-tracking/output index 3c90b8960c..c20efb9801 100644 --- a/testing/btest/Baseline/opt.ZAM-bif-tracking/output +++ b/testing/btest/Baseline/opt.ZAM-bif-tracking/output @@ -1,2 +1,2 @@ ### BTest baseline data generated by btest-diff. Do not edit. Use "btest -U/-u" to update. Requires BTest >= 0.63. -557 seen BiFs, 0 unseen BiFs (), 0 new BiFs () +558 seen BiFs, 0 unseen BiFs (), 0 new BiFs () diff --git a/testing/btest/broker/peering-directionality.zeek b/testing/btest/broker/peering-directionality.zeek new file mode 100644 index 0000000000..080ee7224b --- /dev/null +++ b/testing/btest/broker/peering-directionality.zeek @@ -0,0 +1,54 @@ +# This tests whether the script-layer can correctly query if a given Broker +# peering originated from the local node or from another node that peered with it. +# +# @TEST-GROUP: broker +# @TEST-PORT: BROKER_PORT +# +# @TEST-EXEC: btest-bg-run client "zeek -b ../client.zeek >out" +# @TEST-EXEC: btest-bg-run server "zeek -b ../server.zeek >out" +# +# @TEST-EXEC: btest-bg-wait 15 +# @TEST-EXEC: btest-diff client/out +# @TEST-EXEC: btest-diff server/out + +# @TEST-START-FILE client.zeek +redef exit_only_after_terminate = T; + +event zeek_init() + { + Broker::subscribe("zeek/event/my_topic"); + Broker::peer("127.0.0.1", to_port(getenv("BROKER_PORT"))); + } + +event Broker::peer_added(ep: Broker::EndpointInfo, msg: string) + { + print fmt("peered, this is the outgoing peering: %s", + Broker::is_outbound_peering(ep$network$address, ep$network$bound_port)); + print fmt("via Broker::peers(): %s", Broker::peers()[0]$is_outbound); + + Broker::unpeer("127.0.0.1", to_port(getenv("BROKER_PORT"))); + + print fmt("after unpeering: %s", + Broker::is_outbound_peering(ep$network$address, ep$network$bound_port)); + terminate(); + } +# @TEST-END-FILE + + +# @TEST-START-FILE server.zeek +redef exit_only_after_terminate = T; + +event zeek_init() + { + Broker::subscribe("zeek/event/my_topic"); + Broker::listen("127.0.0.1", to_port(getenv("BROKER_PORT"))); +} + +event Broker::peer_added(ep: Broker::EndpointInfo, msg: string) + { + print fmt("peered, this is the outgoing peering: %s", + Broker::is_outbound_peering(ep$network$address, ep$network$bound_port)); + print fmt("via Broker::peers(): %s", Broker::peers()[0]$is_outbound); + terminate(); + } +# @TEST-END-FILE diff --git a/testing/btest/opt/ZAM-bif-tracking.zeek b/testing/btest/opt/ZAM-bif-tracking.zeek index b57e55e762..4dbce8e247 100644 --- a/testing/btest/opt/ZAM-bif-tracking.zeek +++ b/testing/btest/opt/ZAM-bif-tracking.zeek @@ -38,6 +38,7 @@ global known_BiFs = set( "Broker::__insert_into_set", "Broker::__insert_into_table", "Broker::__is_closed", + "Broker::__is_outbound_peering", "Broker::__keys", "Broker::__listen", "Broker::__node_id",