Add encrypted field to all protocol data tables (#1866)

Summary: Add encrypted field to all protocol data tables

The goal of this column is to ease the ability of identifying
connections that are plaintext that are ingressing from or egressing to
the Internet.

Relevant Issues: #1865

Type of change: /kind feature

Test Plan: Additional test logic verifies new behavior

Changelog Message: Add `encrypted` boolean field to all protocol/L7 data
tables

Signed-off-by: Alma Pixie <[email protected]>

src/stirling/source_connectors/socket_tracer/amqp_table.h

@@ -36,6 +36,7 @@ static constexpr DataElement kAMQPElements[] = {
  canonical_data_elements::kLocalAddr,
  canonical_data_elements::kLocalPort,
  canonical_data_elements::kTraceRole,
+ canonical_data_elements::kEncrypted,
  {
  "frame_type", "AMQP request command",
  types::DataType::INT64,

src/stirling/source_connectors/socket_tracer/canonical_types.h

@@ -78,6 +78,14 @@ constexpr DataElement kLatencyNS = {
  types::PatternType::METRIC_GAUGE,
 };
 
+constexpr DataElement kEncrypted = {
+ "encrypted",
+ "If the protocol trace happened over an encrypted connection",
+ types::DataType::BOOLEAN,
+ types::SemanticType::ST_NONE,
+ types::PatternType::GENERAL_ENUM,
+};
+
 constexpr DataElement kPXInfo = {
  "px_info_",
  "Pixie messages regarding the record (e.g. warnings)",

src/stirling/source_connectors/socket_tracer/cass_table.h

@@ -42,6 +42,7 @@ static constexpr DataElement kCQLElements[] = {
  canonical_data_elements::kLocalAddr,
  canonical_data_elements::kLocalPort,
  canonical_data_elements::kTraceRole,
+ canonical_data_elements::kEncrypted,
  {"req_op", "Request opcode",
  types::DataType::INT64,
  types::SemanticType::ST_NONE,

src/stirling/source_connectors/socket_tracer/dns_table.h

@@ -37,6 +37,7 @@ static constexpr DataElement kDNSElements[] = {
  canonical_data_elements::kLocalAddr,
  canonical_data_elements::kLocalPort,
  canonical_data_elements::kTraceRole,
+ canonical_data_elements::kEncrypted,
  {"req_header", "Request header",
  types::DataType::STRING,
  types::SemanticType::ST_NONE,

src/stirling/source_connectors/socket_tracer/http_table.h

@@ -46,6 +46,7 @@ constexpr DataElement kHTTPElements[] = {
  canonical_data_elements::kLocalAddr,
  canonical_data_elements::kLocalPort,
  canonical_data_elements::kTraceRole,
+ canonical_data_elements::kEncrypted,
  {"major_version", "HTTP major version, can be 1 or 2",
  types::DataType::INT64,
  types::SemanticType::ST_NONE,
@@ -117,6 +118,7 @@ constexpr int kHTTPRemotePortIdx = kHTTPTable.ColIndex("remote_port");
 constexpr int kHTTPLocalAddrIdx = kHTTPTable.ColIndex("local_addr");
 constexpr int kHTTPLocalPortIdx = kHTTPTable.ColIndex("local_port");
 constexpr int kHTTPTraceRoleIdx = kHTTPTable.ColIndex("trace_role");
+constexpr int kHTTPEncryptedIdx = kHTTPTable.ColIndex("encrypted");
 constexpr int kHTTPMajorVersionIdx = kHTTPTable.ColIndex("major_version");
 constexpr int kHTTPMinorVersionIdx = kHTTPTable.ColIndex("minor_version");
 constexpr int kHTTPContentTypeIdx = kHTTPTable.ColIndex("content_type");

src/stirling/source_connectors/socket_tracer/kafka_table.h

@@ -40,6 +40,7 @@ static constexpr DataElement kKafkaElements[] = {
  canonical_data_elements::kLocalAddr,
  canonical_data_elements::kLocalPort,
  canonical_data_elements::kTraceRole,
+ canonical_data_elements::kEncrypted,
  {"req_cmd", "Kafka request command",
  types::DataType::INT64,
  types::SemanticType::ST_NONE,

src/stirling/source_connectors/socket_tracer/mongodb_table.h

@@ -37,6 +37,7 @@ static constexpr DataElement kMongoDBElements[] = {
  canonical_data_elements::kLocalAddr,
  canonical_data_elements::kLocalPort,
  canonical_data_elements::kTraceRole,
+ canonical_data_elements::kEncrypted,
  {"req_cmd", "MongoDB request command",
  types::DataType::STRING,
  types::SemanticType::ST_NONE,

src/stirling/source_connectors/socket_tracer/mux_table.h

@@ -37,6 +37,7 @@ static constexpr DataElement kMuxElements[] = {
  canonical_data_elements::kLocalAddr,
  canonical_data_elements::kLocalPort,
  canonical_data_elements::kTraceRole,
+ canonical_data_elements::kEncrypted,
  {"req_type", "Mux message request type",
  types::DataType::INT64,
  types::SemanticType::ST_NONE,
@@ -59,6 +60,7 @@ DEFINE_PRINT_TABLE(Mux)
 
 static constexpr int kMuxUPIDIdx = kMuxTable.ColIndex("upid");
 static constexpr int kMuxReqTypeIdx = kMuxTable.ColIndex("req_type");
+static constexpr int kMuxEncryptedIdx = kMuxTable.ColIndex("encrypted");
 
 } // namespace stirling
 } // namespace px

src/stirling/source_connectors/socket_tracer/mux_trace_bpf_test.cc

@@ -46,14 +46,13 @@ namespace mux = protocols::mux;
 using ::px::stirling::testing::EqMuxRecord;
 using ::px::stirling::testing::FindRecordIdxMatchesPID;
 using ::px::stirling::testing::FindRecordsMatchingPID;
+using ::px::stirling::testing::GetEncrypted;
 using ::px::stirling::testing::GetTargetRecords;
 using ::px::stirling::testing::SocketTraceBPFTestFixture;
-using ::testing::AllOf;
-using ::testing::UnorderedElementsAre;
-using ::testing::UnorderedElementsAreArray;
 
 using ::testing::Each;
 using ::testing::Field;
+using ::testing::IsFalse;
 using ::testing::MatchesRegex;
 
 // The Init() function is used to set flags for the entire test.
@@ -116,18 +115,6 @@ class MuxTraceTest : public SocketTraceBPFTestFixture</* TClientSideTracing */ t
  ::px::stirling::testing::ThriftMuxServerContainer server_;
 };
 
-std::vector<mux::Record> ToRecordVector(const types::ColumnWrapperRecordBatch& rb,
- const std::vector<size_t>& indices) {
- std::vector<mux::Record> result;
-
- for (const auto& idx : indices) {
- mux::Record r;
- r.req.type = static_cast<int8_t>(rb[kMuxReqTypeIdx]->Get<types::Int64Value>(idx).val);
- result.push_back(r);
- }
- return result;
-}
-
 mux::Record RecordWithType(mux::Type req_type) {
  mux::Record r = {};
  r.req.type = static_cast<int8_t>(req_type);
@@ -148,16 +135,18 @@ TEST_F(MuxTraceTest, Capture) {
 
  // Grab the data from Stirling.
  std::vector<TaggedRecordBatch> tablets = ConsumeRecords(SocketTraceConnector::kMuxTableNum);
- ASSERT_NOT_EMPTY_AND_GET_RECORDS(const types::ColumnWrapperRecordBatch& record_batch, tablets);
+ ASSERT_NOT_EMPTY_AND_GET_RECORDS(const types::ColumnWrapperRecordBatch& rb, tablets);
 
- std::vector<mux::Record> server_records =
- GetTargetRecords<mux::Record>(record_batch, server_.process_pid());
+ const std::vector<size_t> indices =
+ testing::FindRecordIdxMatchesPID(rb, kMuxUPIDIdx, server_.process_pid());
+ std::vector<mux::Record> server_records = testing::ToRecordVector<mux::Record>(rb, indices);
 
  mux::Record tinitCheck = RecordWithType(mux::Type::kRerrOld);
  mux::Record tinit = RecordWithType(mux::Type::kTinit);
  mux::Record pingRecord = RecordWithType(mux::Type::kTping);
  mux::Record dispatchRecord = RecordWithType(mux::Type::kTdispatch);
 
+ EXPECT_THAT(GetEncrypted(rb, kMuxEncryptedIdx, indices), Contains(IsFalse()));
  EXPECT_THAT(server_records, Contains(EqMuxRecord(tinitCheck)));
  EXPECT_THAT(server_records, Contains(EqMuxRecord(tinit)));
  EXPECT_THAT(server_records, Contains(EqMuxRecord(pingRecord)));

src/stirling/source_connectors/socket_tracer/mysql_table.h

@@ -42,6 +42,7 @@ static constexpr DataElement kMySQLElements[] = {
  canonical_data_elements::kLocalAddr,
  canonical_data_elements::kLocalPort,
  canonical_data_elements::kTraceRole,
+ canonical_data_elements::kEncrypted,
  {"req_cmd", "MySQL request command",
  types::DataType::INT64,
  types::SemanticType::ST_NONE,

src/stirling/source_connectors/socket_tracer/nats_table.h

@@ -34,6 +34,7 @@ constexpr DataElement kNATSElements[] = {
  canonical_data_elements::kLocalAddr,
  canonical_data_elements::kLocalPort,
  canonical_data_elements::kTraceRole,
+ canonical_data_elements::kEncrypted,
  {"cmd", "The name of the command.",
  types::DataType::STRING, types::SemanticType::ST_NONE, types::PatternType::GENERAL},
  // For PUB, MSG commands, the parameters and the paylod are included in the 'body' as

src/stirling/source_connectors/socket_tracer/netty_tls_trace_bpf_test.cc

@@ -39,14 +39,14 @@ namespace stirling {
 
 namespace mux = protocols::mux;
 
-using ::px::stirling::testing::EqHTTPRecord;
 using ::px::stirling::testing::EqMuxRecord;
+using ::px::stirling::testing::GetEncrypted;
 using ::px::stirling::testing::GetTargetRecords;
 using ::px::stirling::testing::SocketTraceBPFTestFixture;
 using ::px::stirling::testing::ToRecordVector;
 
+using ::testing::IsTrue;
 using ::testing::StrEq;
-using ::testing::UnorderedElementsAre;
 
 class ThriftMuxServerContainerWrapper : public ::px::stirling::testing::ThriftMuxServerContainer {};
 
@@ -153,15 +153,18 @@ TYPED_TEST(NettyTLSTraceTest, mtls_thriftmux_client) {
  this->StopTransferDataThread();
 
  std::vector<TaggedRecordBatch> tablets = this->ConsumeRecords(SocketTraceConnector::kMuxTableNum);
- ASSERT_NOT_EMPTY_AND_GET_RECORDS(const types::ColumnWrapperRecordBatch& record_batch, tablets);
- std::vector<mux::Record> server_records =
- GetTargetRecords<mux::Record>(record_batch, this->server_.process_pid());
+ ASSERT_NOT_EMPTY_AND_GET_RECORDS(const types::ColumnWrapperRecordBatch& rb, tablets);
+
+ const std::vector<size_t> indices =
+ testing::FindRecordIdxMatchesPID(rb, kMuxUPIDIdx, this->server_.process_pid());
+ std::vector<mux::Record> server_records = ToRecordVector<mux::Record>(rb, indices);
 
  mux::Record tinitCheck = RecordWithType(mux::Type::kRerrOld);
  mux::Record tinit = RecordWithType(mux::Type::kTinit);
  mux::Record pingRecord = RecordWithType(mux::Type::kTping);
  mux::Record dispatchRecord = RecordWithType(mux::Type::kTdispatch);
 
+ EXPECT_THAT(GetEncrypted(rb, kMuxEncryptedIdx, indices), Contains(IsTrue()));
  EXPECT_THAT(server_records, Contains(EqMuxRecord(tinitCheck)));
  EXPECT_THAT(server_records, Contains(EqMuxRecord(tinit)));
  EXPECT_THAT(server_records, Contains(EqMuxRecord(pingRecord)));

src/stirling/source_connectors/socket_tracer/openssl_trace_bpf_test.cc

@@ -52,6 +52,7 @@ using ::px::stirling::testing::GetTargetRecords;
 using ::px::stirling::testing::SocketTraceBPFTestFixture;
 using ::px::stirling::testing::ToRecordVector;
 
+using ::testing::IsTrue;
 using ::testing::StrEq;
 using ::testing::Types;
 using ::testing::UnorderedElementsAre;
@@ -96,6 +97,7 @@ struct TraceRecords {
  std::vector<http::Record> http_records;
  std::vector<std::string> remote_address;
  std::vector<std::string> local_address;
+ std::vector<bool> encrypted;
 };
 
 template <typename TServerContainer>
@@ -127,9 +129,12 @@ class OpenSSLTraceTest : public SocketTraceBPFTestFixture</* TClientSideTracing
  ToRecordVector<http::Record>(record_batch, server_record_indices);
  std::vector<std::string> remote_addresses =
  testing::GetRemoteAddrs(record_batch, server_record_indices);
- std::vector<std::string> local_address =
+ std::vector<std::string> local_addresses =
  testing::GetLocalAddrs(record_batch, server_record_indices);
- return {std::move(http_records), std::move(remote_addresses), std::move(local_address)};
+ std::vector<bool> encrypted =
+ testing::GetEncrypted(record_batch, kHTTPEncryptedIdx, server_record_indices);
+ return {std::move(http_records), std::move(remote_addresses), std::move(local_addresses),
+ std::move(encrypted)};
  }
 
  TServerContainer server_;
@@ -205,6 +210,7 @@ TYPED_TEST(OpenSSLTraceTest, ssl_capture_curl_client) {
  EXPECT_THAT(records.remote_address, UnorderedElementsAre(StrEq("127.0.0.1")));
  // Due to loopback, the local address is the same as the remote address.
  EXPECT_THAT(records.local_address, UnorderedElementsAre(StrEq("127.0.0.1")));
+ EXPECT_THAT(records.encrypted, UnorderedElementsAre(IsTrue()));
 }
 
 TYPED_TEST(OpenSSLTraceTest, ssl_capture_ruby_client) {
@@ -249,6 +255,7 @@ TYPED_TEST(OpenSSLTraceTest, ssl_capture_ruby_client) {
  EqHTTPRecord(expected_record)));
  EXPECT_THAT(records.remote_address,
  UnorderedElementsAre(StrEq("127.0.0.1"), StrEq("127.0.0.1"), StrEq("127.0.0.1")));
+ EXPECT_THAT(records.encrypted, UnorderedElementsAre(IsTrue(), IsTrue(), IsTrue()));
 }
 
 TYPED_TEST(OpenSSLTraceTest, ssl_capture_node_client) {
@@ -269,6 +276,7 @@ TYPED_TEST(OpenSSLTraceTest, ssl_capture_node_client) {
 
  EXPECT_THAT(records.http_records, UnorderedElementsAre(EqHTTPRecord(expected_record)));
  EXPECT_THAT(records.remote_address, UnorderedElementsAre(StrEq("127.0.0.1")));
+ EXPECT_THAT(records.encrypted, UnorderedElementsAre(IsTrue()));
 }
 
 } // namespace stirling

src/stirling/source_connectors/socket_tracer/pgsql_table.h

@@ -34,6 +34,7 @@ static constexpr DataElement kPGSQLElements[] = {
  canonical_data_elements::kLocalAddr,
  canonical_data_elements::kLocalPort,
  canonical_data_elements::kTraceRole,
+ canonical_data_elements::kEncrypted,
  {"req_cmd", "PostgreSQL request command code",
  types::DataType::STRING,
  types::SemanticType::ST_NONE,

src/stirling/source_connectors/socket_tracer/redis_table.h

@@ -37,6 +37,7 @@ static constexpr DataElement kRedisElements[] = {
  canonical_data_elements::kLocalAddr,
  canonical_data_elements::kLocalPort,
  canonical_data_elements::kTraceRole,
+ canonical_data_elements::kEncrypted,
  {"req_cmd", "Request command. See https://redis.io/commands.",
  types::DataType::STRING,
  types::SemanticType::ST_NONE,

src/stirling/source_connectors/socket_tracer/socket_trace_connector.cc

@@ -1275,6 +1275,7 @@ void SocketTraceConnector::AppendMessage(ConnectorContext* ctx, const ConnTracke
  r.Append<r.ColIndex("local_addr")>(conn_tracker.local_endpoint().AddrStr());
  r.Append<r.ColIndex("local_port")>(conn_tracker.local_endpoint().port());
  r.Append<r.ColIndex("trace_role")>(conn_tracker.role());
+ r.Append<r.ColIndex("encrypted")>(conn_tracker.ssl());
  r.Append<r.ColIndex("major_version")>(1);
  r.Append<r.ColIndex("minor_version")>(resp_message.minor_version);
  r.Append<r.ColIndex("content_type")>(static_cast<uint64_t>(content_type));
@@ -1339,6 +1340,7 @@ void SocketTraceConnector::AppendMessage(ConnectorContext* ctx, const ConnTracke
  r.Append<r.ColIndex("local_addr")>(conn_tracker.local_endpoint().AddrStr());
  r.Append<r.ColIndex("local_port")>(conn_tracker.local_endpoint().port());
  r.Append<r.ColIndex("trace_role")>(conn_tracker.role());
+ r.Append<r.ColIndex("encrypted")>(conn_tracker.ssl());
  r.Append<r.ColIndex("major_version")>(2);
  // HTTP2 does not define minor version.
  r.Append<r.ColIndex("minor_version")>(0);
@@ -1382,6 +1384,7 @@ void SocketTraceConnector::AppendMessage(ConnectorContext* ctx, const ConnTracke
  r.Append<r.ColIndex("local_addr")>(conn_tracker.local_endpoint().AddrStr());
  r.Append<r.ColIndex("local_port")>(conn_tracker.local_endpoint().port());
  r.Append<r.ColIndex("trace_role")>(conn_tracker.role());
+ r.Append<r.ColIndex("encrypted")>(conn_tracker.ssl());
  r.Append<r.ColIndex("req_cmd")>(static_cast<uint64_t>(entry.req.cmd));
  r.Append<r.ColIndex("req_body")>(std::move(entry.req.msg), FLAGS_max_body_bytes);
  r.Append<r.ColIndex("resp_status")>(static_cast<uint64_t>(entry.resp.status));
@@ -1407,6 +1410,7 @@ void SocketTraceConnector::AppendMessage(ConnectorContext* ctx, const ConnTracke
  r.Append<r.ColIndex("local_addr")>(conn_tracker.local_endpoint().AddrStr());
  r.Append<r.ColIndex("local_port")>(conn_tracker.local_endpoint().port());
  r.Append<r.ColIndex("trace_role")>(conn_tracker.role());
+ r.Append<r.ColIndex("encrypted")>(conn_tracker.ssl());
  r.Append<r.ColIndex("req_op")>(static_cast<uint64_t>(entry.req.op));
  r.Append<r.ColIndex("req_body")>(std::move(entry.req.msg), FLAGS_max_body_bytes);
  r.Append<r.ColIndex("resp_op")>(static_cast<uint64_t>(entry.resp.op));
@@ -1432,6 +1436,7 @@ void SocketTraceConnector::AppendMessage(ConnectorContext* ctx, const ConnTracke
  r.Append<r.ColIndex("local_addr")>(conn_tracker.local_endpoint().AddrStr());
  r.Append<r.ColIndex("local_port")>(conn_tracker.local_endpoint().port());
  r.Append<r.ColIndex("trace_role")>(conn_tracker.role());
+ r.Append<r.ColIndex("encrypted")>(conn_tracker.ssl());
  r.Append<r.ColIndex("req_header")>(entry.req.header);
  r.Append<r.ColIndex("req_body")>(entry.req.query);
  r.Append<r.ColIndex("resp_header")>(entry.resp.header);
@@ -1457,6 +1462,7 @@ void SocketTraceConnector::AppendMessage(ConnectorContext* ctx, const ConnTracke
  r.Append<r.ColIndex("local_addr")>(conn_tracker.local_endpoint().AddrStr());
  r.Append<r.ColIndex("local_port")>(conn_tracker.local_endpoint().port());
  r.Append<r.ColIndex("trace_role")>(conn_tracker.role());
+ r.Append<r.ColIndex("encrypted")>(conn_tracker.ssl());
  r.Append<r.ColIndex("req")>(std::move(entry.req.payload));
  r.Append<r.ColIndex("resp")>(std::move(entry.resp.payload));
  r.Append<r.ColIndex("latency")>(
@@ -1481,6 +1487,7 @@ void SocketTraceConnector::AppendMessage(ConnectorContext* ctx, const ConnTracke
  r.Append<r.ColIndex("local_addr")>(conn_tracker.local_endpoint().AddrStr());
  r.Append<r.ColIndex("local_port")>(conn_tracker.local_endpoint().port());
  r.Append<r.ColIndex("trace_role")>(conn_tracker.role());
+ r.Append<r.ColIndex("encrypted")>(conn_tracker.ssl());
  r.Append<r.ColIndex("req_type")>(entry.req.type);
  r.Append<r.ColIndex("latency")>(
  CalculateLatency(entry.req.timestamp_ns, entry.resp.timestamp_ns));
@@ -1504,6 +1511,7 @@ void SocketTraceConnector::AppendMessage(ConnectorContext* ctx, const ConnTracke
  r.Append<r.ColIndex("local_addr")>(conn_tracker.local_endpoint().AddrStr());
  r.Append<r.ColIndex("local_port")>(conn_tracker.local_endpoint().port());
  r.Append<r.ColIndex("trace_role")>(conn_tracker.role());
+ r.Append<r.ColIndex("encrypted")>(conn_tracker.ssl());
 
  size_t frame_type = std::max(entry.req.frame_type, entry.resp.frame_type);
  r.Append<r.ColIndex("frame_type")>(frame_type);
@@ -1557,6 +1565,7 @@ void SocketTraceConnector::AppendMessage(ConnectorContext* ctx, const ConnTracke
  r.Append<r.ColIndex("local_addr")>(conn_tracker.local_endpoint().AddrStr());
  r.Append<r.ColIndex("local_port")>(conn_tracker.local_endpoint().port());
  r.Append<r.ColIndex("trace_role")>(role);
+ r.Append<r.ColIndex("encrypted")>(conn_tracker.ssl());
  r.Append<r.ColIndex("req_cmd")>(std::string(entry.req.command));
  r.Append<r.ColIndex("req_args")>(std::string(entry.req.payload));
  r.Append<r.ColIndex("resp")>(std::string(entry.resp.payload));
@@ -1582,6 +1591,7 @@ void SocketTraceConnector::AppendMessage(ConnectorContext* ctx, const ConnTracke
  r.Append<r.ColIndex("local_addr")>(conn_tracker.local_endpoint().AddrStr());
  r.Append<r.ColIndex("local_port")>(conn_tracker.local_endpoint().port());
  r.Append<r.ColIndex("trace_role")>(role);
+ r.Append<r.ColIndex("encrypted")>(conn_tracker.ssl());
  r.Append<r.ColIndex("cmd")>(record.req.command);
  r.Append<r.ColIndex("body")>(record.req.options);
  r.Append<r.ColIndex("resp")>(record.resp.command);
@@ -1607,6 +1617,7 @@ void SocketTraceConnector::AppendMessage(ConnectorContext* ctx, const ConnTracke
  r.Append<r.ColIndex("local_addr")>(conn_tracker.local_endpoint().AddrStr());
  r.Append<r.ColIndex("local_port")>(conn_tracker.local_endpoint().port());
  r.Append<r.ColIndex("trace_role")>(role);
+ r.Append<r.ColIndex("encrypted")>(conn_tracker.ssl());
  r.Append<r.ColIndex("req_cmd")>(static_cast<int64_t>(record.req.api_key));
  r.Append<r.ColIndex("client_id")>(std::move(record.req.client_id), FLAGS_max_body_bytes);
  r.Append<r.ColIndex("req_body")>(std::move(record.req.msg), kMaxKafkaBodyBytes);
@@ -1633,6 +1644,7 @@ void SocketTraceConnector::AppendMessage(ConnectorContext* ctx, const ConnTracke
  r.Append<r.ColIndex("local_addr")>(conn_tracker.local_endpoint().AddrStr());
  r.Append<r.ColIndex("local_port")>(conn_tracker.local_endpoint().port());
  r.Append<r.ColIndex("trace_role")>(role);
+ r.Append<r.ColIndex("encrypted")>(conn_tracker.ssl());
  r.Append<r.ColIndex("req_cmd")>(std::move(record.req.op_msg_type));
  r.Append<r.ColIndex("req_body")>(std::move(record.req.frame_body));
  r.Append<r.ColIndex("resp_status")>(std::move(record.resp.op_msg_type));

src/stirling/source_connectors/socket_tracer/testing/protocol_checkers.h

@@ -126,6 +126,15 @@ inline std::vector<std::string> GetLocalAddrs(const types::ColumnWrapperRecordBa
  return addrs;
 }
 
+inline std::vector<bool> GetEncrypted(const types::ColumnWrapperRecordBatch& rb,
+ const int encrypted_idx, const std::vector<size_t>& indices) {
+ std::vector<bool> encrypted;
+ for (size_t idx : indices) {
+ encrypted.push_back(rb[encrypted_idx]->Get<types::BoolValue>(idx).val);
+ }
+ return encrypted;
+}
+
 inline std::vector<int64_t> GetRemotePorts(const types::ColumnWrapperRecordBatch& rb,
  const std::vector<size_t>& indices) {
  std::vector<int64_t> addrs;