ROX-29399: Directional external-IPs

collector/lib/CollectorConfig.cpp

@@ -91,7 +91,6 @@ constexpr const char* CollectorConfig::kSyscalls[];
 constexpr bool CollectorConfig::kEnableProcessesListeningOnPorts;
 
 const UnorderedSet<L4ProtoPortPair> CollectorConfig::kIgnoredL4ProtoPortPairs = {{L4Proto::UDP, 9}};
-;
 
 CollectorConfig::CollectorConfig() {
   // Set default configuration values
@@ -439,7 +438,7 @@ std::ostream& operator<<(std::ostream& os, const CollectorConfig& c) {
          << ", set_import_users:" << c.ImportUsers()
          << ", collect_connection_status:" << c.CollectConnectionStatus()
          << ", enable_detailed_metrics:" << c.EnableDetailedMetrics()
-         << ", enable_external_ips:" << c.EnableExternalIPs()
+         << ", external_ips:" << c.ExternalIPsConf()
          << ", track_send_recv:" << c.TrackingSendRecv();
 }
 
@@ -545,4 +544,38 @@ std::pair<option::ArgStatus, std::string> CollectorConfig::CheckConfiguration(co
   return std::make_pair(ARG_OK, "");
 }
 
+CollectorConfig::ExternalIPsConfig::ExternalIPsConfig(std::optional<sensor::CollectorConfig> runtime_config, bool default_enabled)
+    : direction_enabled_(Direction::NONE) {
+  if (runtime_config.has_value()) {
+    if (runtime_config.value().networking().external_ips().enabled() == sensor::ExternalIpsEnabled::ENABLED) {
+      switch (runtime_config.value().networking().external_ips().direction()) {
+        case sensor::ExternalIpsDirection::INGRESS:
+          direction_enabled_ = Direction::INGRESS;
+          break;
+        case sensor::ExternalIpsDirection::EGRESS:
+          direction_enabled_ = Direction::EGRESS;
+          break;
+        default:
+          direction_enabled_ = Direction::BOTH;
+          break;
+      }
+    }
+  } else if (default_enabled) {
+    direction_enabled_ = Direction::BOTH;
+  }
+}
+
+std::ostream& operator<<(std::ostream& os, const collector::CollectorConfig::ExternalIPsConfig& config) {
+  static std::unordered_map<collector::CollectorConfig::ExternalIPsConfig::Direction, std::string> mapping = {
+      {CollectorConfig::ExternalIPsConfig::Direction::NONE, "NONE"},
+      {CollectorConfig::ExternalIPsConfig::Direction::INGRESS, "INGRESS"},
+      {CollectorConfig::ExternalIPsConfig::Direction::EGRESS, "EGRESS"},
+      {CollectorConfig::ExternalIPsConfig::Direction::BOTH, "BOTH"},
+  };
+
+  auto str = mapping.find(config.GetDirection());
+
+  return os << "direction(" << ((str == mapping.end()) ? "invalid" : (*str).second) << ")";
+}
+
 }  // namespace collector

collector/lib/CollectorConfig.h

@@ -94,18 +94,39 @@ class CollectorConfig {
   bool ImportUsers() const { return import_users_; }
   bool CollectConnectionStatus() const { return collect_connection_status_; }
 
+  // Encapsulates the configuration of the External-IPs feature
+  class ExternalIPsConfig {
+   public:
+    enum Direction {
+      NONE = 0,
+      INGRESS = 1 << 0,
+      EGRESS = 1 << 1,
+      BOTH = INGRESS | EGRESS,
+    };
+
+    // Are External-IPs enabled in the provided direction ?
+    bool IsEnabled(Direction direction) { return (direction & direction_enabled_) == direction; }
+
+    // Direction in which External-IPs are enabled
+    Direction GetDirection() const { return direction_enabled_; }
+
+    // Extract the External-IPs configuration from the provided runtime-conf.
+    // If the runtime-configuration is unset then 'default_enabled' is used
+    // as a fallback to enable in both directions.
+    // 'runtime_config' should be locked prior to calling.
+    ExternalIPsConfig(std::optional<sensor::CollectorConfig> runtime_config, bool default_enabled);
+
+   private:
+    Direction direction_enabled_;
+  };
+
   // EnableExternalIPs will check for the existence
   // of a runtime configuration, and defer to that value
   // otherwise, we rely on the feature flag (env var)
-  bool EnableExternalIPs() const {
+  ExternalIPsConfig ExternalIPsConf() const {
     auto lock = ReadLock();
-    if (runtime_config_.has_value()) {
-      return runtime_config_.value()
-                 .networking()
-                 .external_ips()
-                 .enabled() == sensor::ExternalIpsEnabled::ENABLED;
-    }
-    return enable_external_ips_;
+
+    return ExternalIPsConfig(runtime_config_, enable_external_ips_);
   }
 
   void RuntimeConfigHeuristics() {
@@ -266,5 +287,6 @@ class CollectorConfig {
 };
 
 std::ostream& operator<<(std::ostream& os, const CollectorConfig& c);
+std::ostream& operator<<(std::ostream& os, const collector::CollectorConfig::ExternalIPsConfig& config);
 
 }  // end namespace collector

collector/lib/ConnTracker.cpp

@@ -112,7 +112,7 @@ IPNet ConnectionTracker::NormalizeAddressNoLock(const Address& address, bool ena
 }
 
 bool ConnectionTracker::ShouldNormalizeConnection(const Connection* conn) const {
-  Endpoint local, remote = conn->remote();
+  Endpoint remote = conn->remote();
   IPNet ipnet = NormalizeAddressNoLock(remote.address(), false);
 
   return Address::IsCanonicalExternalIp(ipnet.address());
@@ -138,28 +138,35 @@ void ConnectionTracker::CloseConnections(ConnMap* old_conn_state, ConnMap* delta
 
 /**
  * Closes connections that have the 255.255.255.255 external IP address
+ * Affects only connections with a matching is_server property
  */
-void ConnectionTracker::CloseNormalizedConnections(ConnMap* old_conn_state, ConnMap* delta_conn) {
-  CloseConnections(old_conn_state, delta_conn, [](const Connection* conn) {
-    return Address::IsCanonicalExternalIp(conn->remote().address());
+void ConnectionTracker::CloseNormalizedConnections(bool is_server, ConnMap* old_conn_state, ConnMap* delta_conn) {
+  CloseConnections(old_conn_state, delta_conn, [is_server](const Connection* conn) {
+    return conn->is_server() == is_server && Address::IsCanonicalExternalIp(conn->remote().address());
   });
 }
 
 /**
  * Closes unnormalized connections that would be normalized to the canonical external
  * IP address if external IPs was enabled
+ * Affects only connections with a matching is_server property
  */
-void ConnectionTracker::CloseExternalUnnormalizedConnections(ConnMap* old_conn_state, ConnMap* delta_conn) {
-  CloseConnections(old_conn_state, delta_conn, [this](const Connection* conn) {
-    return ShouldNormalizeConnection(conn) && !Address::IsCanonicalExternalIp(conn->remote().address());
+void ConnectionTracker::CloseExternalUnnormalizedConnections(bool is_server, ConnMap* old_conn_state, ConnMap* delta_conn) {
+  CloseConnections(old_conn_state, delta_conn, [this, is_server](const Connection* conn) {
+    return conn->is_server() == is_server && ShouldNormalizeConnection(conn) && !Address::IsCanonicalExternalIp(conn->remote().address());
   });
 }
 
-void ConnectionTracker::CloseConnectionsOnRuntimeConfigChange(ConnMap* old_conn_state, ConnMap* delta_conn, bool enableExternalIPs) {
-  if (enableExternalIPs) {
-    CloseNormalizedConnections(old_conn_state, delta_conn);
+void ConnectionTracker::CloseConnectionsOnRuntimeConfigChange(ConnMap* old_conn_state, ConnMap* delta_conn) {
+  if (enable_external_ips_egress_) {
+    CloseNormalizedConnections(/* egress is when we are not server */ false, old_conn_state, delta_conn);
   } else {
-    CloseExternalUnnormalizedConnections(old_conn_state, delta_conn);
+    CloseExternalUnnormalizedConnections(/* egress is when we are not server */ false, old_conn_state, delta_conn);
+  }
+  if (enable_external_ips_ingress_) {
+    CloseNormalizedConnections(/* ingress is when we are server */ true, old_conn_state, delta_conn);
+  } else {
+    CloseExternalUnnormalizedConnections(/* ingress is when we are server */ true, old_conn_state, delta_conn);
   }
 }
 
@@ -175,11 +182,11 @@ Connection ConnectionTracker::NormalizeConnectionNoLock(const Connection& conn)
   if (is_server) {
     // If this is the server, only the local port is relevant, while the remote port does not matter.
     local = Endpoint(IPNet(Address()), conn.local().port());
-    remote = Endpoint(NormalizeAddressNoLock(conn.remote().address(), enable_external_ips_), 0);
+    remote = Endpoint(NormalizeAddressNoLock(conn.remote().address(), enable_external_ips_ingress_), 0);
   } else {
     // If this is the client, the local port and address are not relevant.
     local = Endpoint();
-    remote = Endpoint(NormalizeAddressNoLock(remote.address(), enable_external_ips_), remote.port());
+    remote = Endpoint(NormalizeAddressNoLock(remote.address(), enable_external_ips_egress_), remote.port());
   }
 
   return Connection(conn.container(), local, remote, conn.l4proto(), is_server);

collector/lib/ConnTracker.h

@@ -101,9 +101,9 @@ class ConnectionTracker {
   static void UpdateOldState(UnorderedMap<T, ConnStatus>* old_state, const UnorderedMap<T, ConnStatus>& new_state, int64_t time_micros, int64_t afterglow_period_micros);
 
   void CloseConnections(ConnMap* old_conn_state, ConnMap* delta_conn, std::function<bool(const Connection*)> predicate);
-  void CloseNormalizedConnections(ConnMap* old_conn_state, ConnMap* delta_conn);
-  void CloseExternalUnnormalizedConnections(ConnMap* old_conn_state, ConnMap* delta_conn);
-  void CloseConnectionsOnRuntimeConfigChange(ConnMap* old_conn_state, ConnMap* delta_conn, bool enableExternalIPs);
+  void CloseNormalizedConnections(bool is_server, ConnMap* old_conn_state, ConnMap* delta_conn);
+  void CloseExternalUnnormalizedConnections(bool is_server, ConnMap* old_conn_state, ConnMap* delta_conn);
+  void CloseConnectionsOnRuntimeConfigChange(ConnMap* old_conn_state, ConnMap* delta_conn);
 
   // ComputeDelta computes a diff between new_state and old_state
   template <typename T>
@@ -131,7 +131,10 @@ class ConnectionTracker {
 
   void UpdateKnownPublicIPs(UnorderedSet<Address>&& known_public_ips);
   void UpdateKnownIPNetworks(UnorderedMap<Address::Family, std::vector<IPNet>>&& known_ip_networks);
-  void EnableExternalIPs(bool enable) { enable_external_ips_ = enable; }
+  void EnableExternalIPs(bool enable_ingress, bool enable_egress) {
+    enable_external_ips_ingress_ = enable_ingress;
+    enable_external_ips_egress_ = enable_egress;
+  }
   void UpdateIgnoredL4ProtoPortPairs(UnorderedSet<L4ProtoPortPair>&& ignored_l4proto_port_pairs);
   void UpdateIgnoredNetworks(const std::vector<IPNet>& network_list);
   void UpdateNonAggregatedNetworks(const std::vector<IPNet>& network_list);
@@ -202,7 +205,8 @@ class ConnectionTracker {
 
   UnorderedSet<Address> known_public_ips_;
   NRadixTree known_ip_networks_;
-  bool enable_external_ips_ = false;
+  bool enable_external_ips_ingress_ = false;
+  bool enable_external_ips_egress_ = false;
   UnorderedMap<Address::Family, bool> known_private_networks_exists_;
   UnorderedSet<L4ProtoPortPair> ignored_l4proto_port_pairs_;
   NRadixTree ignored_networks_;

collector/lib/NetworkStatusNotifier.cpp

@@ -16,6 +16,8 @@ namespace collector {
 
 namespace {
 
+using Direction = CollectorConfig::ExternalIPsConfig::Direction;
+
 storage::L4Protocol TranslateL4Protocol(L4Proto proto) {
   switch (proto) {
     case L4Proto::TCP:
@@ -225,7 +227,7 @@ void NetworkStatusNotifier::RunSingle(IDuplexClientWriter<sensor::NetworkConnect
   auto next_scrape = std::chrono::system_clock::now();
   int64_t time_at_last_scrape = NowMicros();
 
-  bool prevEnableExternalIPs = config_.EnableExternalIPs();
+  CollectorConfig::ExternalIPsConfig prevEnableExternalIPs = config_.ExternalIPsConf();
 
   while (writer->Sleep(next_scrape)) {
     CLOG(DEBUG) << "Starting network status notification";
@@ -242,17 +244,19 @@ void NetworkStatusNotifier::RunSingle(IDuplexClientWriter<sensor::NetworkConnect
     const sensor::NetworkConnectionInfoMessage* msg;
     ConnMap new_conn_state, delta_conn;
     AdvertisedEndpointMap new_cep_state;
-    bool enableExternalIPs = config_.EnableExternalIPs();
+    CollectorConfig::ExternalIPsConfig externalIPsConfig = config_.ExternalIPsConf();
 
     WITH_TIMER(CollectorStats::net_fetch_state) {
-      conn_tracker_->EnableExternalIPs(enableExternalIPs);
+      conn_tracker_->EnableExternalIPs(
+          externalIPsConfig.IsEnabled(Direction::INGRESS),
+          externalIPsConfig.IsEnabled(Direction::EGRESS));
 
       new_conn_state = conn_tracker_->FetchConnState(true, true);
       if (config_.EnableAfterglow()) {
         ConnectionTracker::ComputeDeltaAfterglow(new_conn_state, old_conn_state, delta_conn, time_micros, time_at_last_scrape, config_.AfterglowPeriod());
-        if (prevEnableExternalIPs != enableExternalIPs) {
-          conn_tracker_->CloseConnectionsOnRuntimeConfigChange(&old_conn_state, &delta_conn, enableExternalIPs);
-          prevEnableExternalIPs = enableExternalIPs;
+        if (prevEnableExternalIPs.GetDirection() != externalIPsConfig.GetDirection()) {
+          conn_tracker_->CloseConnectionsOnRuntimeConfigChange(&old_conn_state, &delta_conn);
+          prevEnableExternalIPs = externalIPsConfig;
         }
       } else {
         ConnectionTracker::ComputeDelta(new_conn_state, &old_conn_state);

collector/test/CollectorConfigTest.cpp

@@ -8,6 +8,7 @@
 #include "gtest/gtest.h"
 
 using namespace testing;
+using Direction = collector::CollectorConfig::ExternalIPsConfig::Direction;
 
 namespace collector {
 
@@ -115,11 +116,11 @@ TEST(CollectorConfigTest, TestEnableExternalIpsFeatureFlag) {
 
   config.MockSetEnableExternalIPs(false);
 
-  EXPECT_FALSE(config.EnableExternalIPs());
+  EXPECT_EQ(Direction::NONE, config.ExternalIPsConf().GetDirection());
 
   config.MockSetEnableExternalIPs(true);
 
-  EXPECT_TRUE(config.EnableExternalIPs());
+  EXPECT_EQ(Direction::BOTH, config.ExternalIPsConf().GetDirection());
 }
 
 TEST(CollectorConfigTest, TestEnableExternalIpsRuntimeConfig) {
@@ -138,14 +139,20 @@ TEST(CollectorConfigTest, TestEnableExternalIpsRuntimeConfig) {
 
   config.SetRuntimeConfig(runtime_config);
 
-  EXPECT_FALSE(config.EnableExternalIPs());
+  EXPECT_EQ(Direction::NONE, config.ExternalIPsConf().GetDirection());
+  EXPECT_FALSE(config.ExternalIPsConf().IsEnabled(Direction::INGRESS));
+  EXPECT_FALSE(config.ExternalIPsConf().IsEnabled(Direction::EGRESS));
+  EXPECT_FALSE(config.ExternalIPsConf().IsEnabled(Direction::BOTH));
 
   config.MockSetEnableExternalIPs(false);
 
   external_ips_config->set_enabled(sensor::ExternalIpsEnabled::ENABLED);
   config.SetRuntimeConfig(runtime_config);
 
-  EXPECT_TRUE(config.EnableExternalIPs());
+  EXPECT_EQ(CollectorConfig::ExternalIPsConfig::Direction::BOTH, config.ExternalIPsConf().GetDirection());
+  EXPECT_TRUE(config.ExternalIPsConf().IsEnabled(Direction::INGRESS));
+  EXPECT_TRUE(config.ExternalIPsConf().IsEnabled(Direction::EGRESS));
+  EXPECT_TRUE(config.ExternalIPsConf().IsEnabled(Direction::BOTH));
 }
 
 }  // namespace collector

collector/test/ConfigLoaderTest.cpp

@@ -11,6 +11,7 @@
 
 namespace collector {
 using namespace google::protobuf::util;
+using Direction = CollectorConfig::ExternalIPsConfig::Direction;
 
 std::string ErrorsToString(const std::vector<ParserError>& errors) {
   std::stringstream ss;
@@ -308,29 +309,49 @@ TEST(TestParserYaml, ValidationMode) {
  */
 
 TEST(CollectorConfigTest, TestYamlConfigToConfigMultiple) {
-  std::vector<std::pair<std::string, bool>> tests = {
+  std::vector<std::pair<std::string, Direction>> tests = {
       {R"(
                   networking:
                     externalIps:
                       enabled: enabled
                )",
-       true},
+       Direction::BOTH},
+      {R"(
+                  networking:
+                    externalIps:
+                      enabled: enabled
+                      direction: ingress
+               )",
+       Direction::INGRESS},
+      {R"(
+                  networking:
+                    externalIps:
+                      enabled: enabled
+                      direction: egress
+               )",
+       Direction::EGRESS},
+      {R"(
+                  networking:
+                    externalIps:
+                      enabled: enabled
+                      direction: both
+               )",
+       Direction::BOTH},
       {R"(
                   networking:
                     externalIps:
                       enabled: DISABLED
                )",
-       false},
+       Direction::NONE},
       {R"(
                   networking:
                     externalIps:
                )",
-       false},
-      {
-          R"(
+       Direction::NONE},
+      {R"(
                   networking:
                )",
-          false},
+       Direction::NONE},
   };
 
   for (const auto& [yamlStr, expected] : tests) {
@@ -342,12 +363,7 @@ TEST(CollectorConfigTest, TestYamlConfigToConfigMultiple) {
 
     EXPECT_TRUE(runtime_config.has_value());
 
-    bool enabled = runtime_config.value()
-                       .networking()
-                       .external_ips()
-                       .enabled() == sensor::ExternalIpsEnabled::ENABLED;
-    EXPECT_EQ(enabled, expected);
-    EXPECT_EQ(config.EnableExternalIPs(), expected);
+    EXPECT_EQ(config.ExternalIPsConf().GetDirection(), expected);
   }
 }