Fix leak in SecurityManager::participant_volatile_message_secure_writ…

…er_ (#4673)

* Refs #20658. Add blackbox test.

Signed-off-by: Miguel Company <miguelcompany@eprosima.com>

* Refs #20658. Add expectations to unit test.

Signed-off-by: Miguel Company <miguelcompany@eprosima.com>

* Refs #20658. Fix issue.

Signed-off-by: Miguel Company <miguelcompany@eprosima.com>

* Refs #20658. Improve regression test.

Signed-off-by: Miguel Company <miguelcompany@eprosima.com>

---------

Signed-off-by: Miguel Company <miguelcompany@eprosima.com>

src/cpp/rtps/security/SecurityManager.cpp

@@ -1263,7 +1263,7 @@ bool SecurityManager::create_participant_volatile_message_secure_writer()
     RTPSWriter* wout = nullptr;
     if (participant_->createWriter(&wout, watt, participant_volatile_message_secure_pool_,
             participant_volatile_message_secure_writer_history_,
-            nullptr, participant_volatile_message_secure_writer_entity_id, true))
+            this, participant_volatile_message_secure_writer_entity_id, true))
     {
         participant_->set_endpoint_rtps_protection_supports(wout, false);
         participant_volatile_message_secure_writer_ = dynamic_cast<StatefulWriter*>(wout);
@@ -4341,3 +4341,16 @@ bool SecurityManager::DiscoveredParticipantInfo::check_guid_comes_from(
     }
     return ret;
 }
+
+void SecurityManager::onWriterChangeReceivedByAll(
+        RTPSWriter* writer,
+        CacheChange_t* change)
+{
+    static_cast<void>(writer);
+    assert(writer == participant_volatile_message_secure_writer_);
+
+    if (nullptr != participant_volatile_message_secure_writer_history_)
+    {
+        participant_volatile_message_secure_writer_history_->remove_change(change);
+    }
+}

src/cpp/rtps/security/SecurityManager.h

@@ -37,6 +37,7 @@
 #include <fastdds/rtps/resources/TimedEvent.h>
 #include <fastdds/rtps/security/authentication/Handshake.h>
 #include <fastdds/rtps/security/common/ParticipantGenericMessage.h>
+#include <fastdds/rtps/writer/WriterListener.h>
 #include <fastrtps/utils/ProxyPool.hpp>
 #include <fastrtps/utils/shared_mutex.hpp>
 
@@ -45,6 +46,7 @@ namespace fastrtps {
 namespace rtps {
 
 class RTPSParticipantImpl;
+class RTPSWriter;
 class StatelessWriter;
 class StatelessReader;
 class StatefulWriter;
@@ -66,7 +68,7 @@ struct EndpointSecurityAttributes;
  *
  * @ingroup SECURITY_MODULE
  */
-class SecurityManager
+class SecurityManager : private WriterListener
 {
 public:
 
@@ -874,6 +876,10 @@ class SecurityManager
         }
     }
 
+    void onWriterChangeReceivedByAll(
+            RTPSWriter* writer,
+            CacheChange_t* change) override;
+
     /**
      * Syncronization object for plugin initialization, <tt>mutex_</tt> protection is not necessary to guarantee plugin
      * availability.

test/blackbox/common/BlackboxTestsSecurity.cpp

@@ -3266,6 +3266,119 @@ TEST_P(Security, BuiltinAuthenticationAndAccessAndCryptoPlugin_Permissions_valid
     }
 }
 
+// Regression test of Refs #20658, Github #4553.
+TEST_P(Security, BuiltinAuthenticationAndAccessAndCryptoPlugin_Permissions_validation_toggle_partition)
+{
+    PubSubWriter<HelloWorldPubSubType> writer("HelloWorldTopic");
+    PubSubReader<HelloWorldPubSubType> reader_p_1("HelloWorldTopic");
+    PubSubReader<HelloWorldPubSubType> reader_p_2("HelloWorldTopic");
+
+    std::string governance_file("governance_helloworld_all_enable.smime");
+
+    // Prepare subscriptions security properties
+    PropertyPolicy sub_property_policy;
+    sub_property_policy.properties().emplace_back(Property("dds.sec.auth.plugin",
+            "builtin.PKI-DH"));
+    sub_property_policy.properties().emplace_back(Property("dds.sec.auth.builtin.PKI-DH.identity_ca",
+            "file://" + std::string(certs_path) + "/maincacert.pem"));
+    sub_property_policy.properties().emplace_back(Property("dds.sec.auth.builtin.PKI-DH.identity_certificate",
+            "file://" + std::string(certs_path) + "/mainsubcert.pem"));
+    sub_property_policy.properties().emplace_back(Property("dds.sec.auth.builtin.PKI-DH.private_key",
+            "file://" + std::string(certs_path) + "/mainsubkey.pem"));
+    sub_property_policy.properties().emplace_back(Property("dds.sec.crypto.plugin",
+            "builtin.AES-GCM-GMAC"));
+    sub_property_policy.properties().emplace_back(Property("dds.sec.access.plugin",
+            "builtin.Access-Permissions"));
+    sub_property_policy.properties().emplace_back(Property(
+                "dds.sec.access.builtin.Access-Permissions.permissions_ca",
+                "file://" + std::string(certs_path) + "/maincacert.pem"));
+    sub_property_policy.properties().emplace_back(Property("dds.sec.access.builtin.Access-Permissions.governance",
+            "file://" + std::string(certs_path) + "/" + governance_file));
+    sub_property_policy.properties().emplace_back(Property("dds.sec.access.builtin.Access-Permissions.permissions",
+            "file://" + std::string(certs_path) + "/permissions_helloworld_partitions.smime"));
+
+    // Initialize one reader on each partition
+    reader_p_1.partition("Partition1").
+            reliability(eprosima::fastrtps::RELIABLE_RELIABILITY_QOS).
+            property_policy(sub_property_policy).
+            init();
+    ASSERT_TRUE(reader_p_1.isInitialized());
+
+    reader_p_2.partition("Partition2").
+            reliability(eprosima::fastrtps::RELIABLE_RELIABILITY_QOS).
+            property_policy(sub_property_policy).
+            init();
+    ASSERT_TRUE(reader_p_2.isInitialized());
+
+    // Prepare publication security properties
+    PropertyPolicy pub_property_policy;
+    pub_property_policy.properties().emplace_back(Property("dds.sec.auth.plugin",
+            "builtin.PKI-DH"));
+    pub_property_policy.properties().emplace_back(Property("dds.sec.auth.builtin.PKI-DH.identity_ca",
+            "file://" + std::string(certs_path) + "/maincacert.pem"));
+    pub_property_policy.properties().emplace_back(Property("dds.sec.auth.builtin.PKI-DH.identity_certificate",
+            "file://" + std::string(certs_path) + "/mainpubcert.pem"));
+    pub_property_policy.properties().emplace_back(Property("dds.sec.auth.builtin.PKI-DH.private_key",
+            "file://" + std::string(certs_path) + "/mainpubkey.pem"));
+    pub_property_policy.properties().emplace_back(Property("dds.sec.crypto.plugin",
+            "builtin.AES-GCM-GMAC"));
+    pub_property_policy.properties().emplace_back(Property("dds.sec.access.plugin",
+            "builtin.Access-Permissions"));
+    pub_property_policy.properties().emplace_back(Property(
+                "dds.sec.access.builtin.Access-Permissions.permissions_ca",
+                "file://" + std::string(certs_path) + "/maincacert.pem"));
+    pub_property_policy.properties().emplace_back(Property("dds.sec.access.builtin.Access-Permissions.governance",
+            "file://" + std::string(certs_path) + "/" + governance_file));
+    pub_property_policy.properties().emplace_back(Property("dds.sec.access.builtin.Access-Permissions.permissions",
+            "file://" + std::string(certs_path) + "/permissions_helloworld_partitions.smime"));
+
+    // Initialize a writer on both partitions
+    writer.partition("Partition1").partition("Partition2").
+            property_policy(pub_property_policy).
+            init();
+    ASSERT_TRUE(writer.isInitialized());
+
+    // Wait for all entities to discover each other
+    reader_p_1.wait_discovery();
+    reader_p_2.wait_discovery();
+    writer.wait_discovery(2u);
+
+    constexpr size_t num_samples = 100;
+    auto data = default_helloworld_data_generator(num_samples);
+    reader_p_1.startReception(data);
+    reader_p_2.startReception(data);
+
+    for (size_t i = 0; i < num_samples; ++i)
+    {
+        // Switch to third partition and wait for all entities to unmatch
+        writer.update_partition("Partition3");
+        reader_p_1.wait_writer_undiscovery();
+        reader_p_2.wait_writer_undiscovery();
+        writer.wait_discovery(0u);
+
+        // Switch partition and wait for the corresponding reader to discover the writer
+        if (0 == i % 2)
+        {
+            writer.update_partition("Partition1");
+            reader_p_1.wait_discovery();
+        }
+        else
+        {
+            writer.update_partition("Partition2");
+            reader_p_2.wait_discovery();
+        }
+
+        // Ensure the writer matches the reader before sending the sample
+        writer.wait_discovery(1u);
+        writer.send_sample(data.front());
+        data.pop_front();
+        writer.waitForAllAcked(std::chrono::milliseconds(100));
+    }
+
+    EXPECT_EQ(num_samples / 2u, reader_p_1.getReceivedCount());
+    EXPECT_EQ(num_samples / 2u, reader_p_2.getReceivedCount());
+}
+
 template <typename DataType>
 void prepare_pkcs11_nodes(
         PubSubReader<DataType>& reader,

test/unittest/rtps/security/SecurityHandshakeProcessTests.cpp

@@ -337,8 +337,13 @@ TEST_F(SecurityTest, discovered_participant_process_message_ok_begin_handshake_r
     info.guid = participant_data.m_guid;
     EXPECT_CALL(*participant_.getListener(), onParticipantAuthentication(_, info)).Times(1);
 
+    CacheChange_t* kx_change = new CacheChange_t(500);
+    expect_kx_exchange(kx_change);
+
     stateless_reader_->listener_->onNewCacheChangeAdded(stateless_reader_, change);
 
+    volatile_writer_->listener_->onWriterChangeReceivedByAll(volatile_writer_, kx_change);
+
     return_handle(remote_identity_handle);
     return_handle(handshake_handle);
 }
@@ -522,10 +527,15 @@ TEST_F(SecurityTest, discovered_participant_process_message_pending_handshake_re
             WillOnce(DoAll(SetArgPointee<0>(&remote_identity_handle),
             Return(ValidationResult_t::VALIDATION_PENDING_HANDSHAKE_MESSAGE)));
 
+    CacheChange_t* kx_change = new CacheChange_t(500);
+    expect_kx_exchange(kx_change);
+
     ParticipantProxyData participant_data;
     fill_participant_key(participant_data.m_guid);
     ASSERT_TRUE(manager_.discovered_participant(participant_data));
 
+    volatile_writer_->listener_->onWriterChangeReceivedByAll(volatile_writer_, kx_change);
+
     ParticipantGenericMessage message;
     message.message_identity().source_guid(participant_data.m_guid);
     message.destination_participant_key(participant_data.m_guid);
@@ -722,7 +732,12 @@ TEST_F(SecurityTest, discovered_participant_process_message_ok_process_handshake
     info.guid = remote_participant_key;
     EXPECT_CALL(*participant_.getListener(), onParticipantAuthentication(_, info)).Times(1);
 
+    CacheChange_t* kx_change = new CacheChange_t(500);
+    expect_kx_exchange(kx_change);
+
     stateless_reader_->listener_->onNewCacheChangeAdded(stateless_reader_, change);
+
+    volatile_writer_->listener_->onWriterChangeReceivedByAll(volatile_writer_, kx_change);
 }
 
 TEST_F(SecurityTest, discovered_participant_process_message_process_handshake_reply_new_change_fail)
@@ -1116,7 +1131,12 @@ TEST_F(SecurityTest, discovered_participant_process_message_ok_process_handshake
     info.guid = remote_participant_key;
     EXPECT_CALL(*participant_.getListener(), onParticipantAuthentication(_, info)).Times(1);
 
+    CacheChange_t* kx_change = new CacheChange_t(500);
+    expect_kx_exchange(kx_change);
+
     stateless_reader_->listener_->onNewCacheChangeAdded(stateless_reader_, change);
+
+    volatile_writer_->listener_->onWriterChangeReceivedByAll(volatile_writer_, kx_change);
 }
 
 int main(

test/unittest/rtps/security/SecurityTests.cpp

@@ -21,7 +21,7 @@ void SecurityTest::initialization_ok()
     ::testing::DefaultValue<const ParticipantSecurityAttributes&>::Set(security_attributes_);
     stateless_writer_ = new ::testing::NiceMock<StatelessWriter>(&participant_);
     stateless_reader_ = new ::testing::NiceMock<StatelessReader>();
-    volatile_writer_ = new ::testing::NiceMock<StatefulWriter>(&participant_);
+    volatile_writer_ = new ::testing::StrictMock<StatefulWriter>(&participant_);
     volatile_reader_ = new ::testing::NiceMock<StatefulReader>();
 
     EXPECT_CALL(*auth_plugin_, validate_local_identity(_, _, _, _, _, _)).Times(1).
@@ -34,15 +34,18 @@ void SecurityTest::initialization_ok()
             WillOnce(Return(true));
     EXPECT_CALL(participant_, createWriter_mock(_, _, _, _, _, _)).Times(2).
             WillOnce(DoAll(SetArgPointee<0>(stateless_writer_), Return(true))).
-            WillOnce(DoAll(SetArgPointee<0>(volatile_writer_), Return(true)));
+            WillOnce(DoAll(SaveArg<3>(&volatile_writer_->listener_), SetArgPointee<0>(volatile_writer_), Return(true)));
     EXPECT_CALL(participant_, createReader_mock(_, _, _, _, _, _, _)).Times(2).
             WillOnce(DoAll(SetArgPointee<0>(stateless_reader_), Return(true))).
             WillOnce(DoAll(SetArgPointee<0>(volatile_reader_), Return(true)));
 
+    EXPECT_CALL(*volatile_writer_, set_separate_sending(true)).Times(1);
+
     security_activated_ = manager_.init(security_attributes_, participant_properties_);
     ASSERT_TRUE(security_activated_);
     ASSERT_TRUE(manager_.is_security_initialized());
     ASSERT_TRUE(manager_.create_entities());
+    ASSERT_TRUE(volatile_writer_->listener_ != nullptr);
 }
 
 void SecurityTest::initialization_auth_ok()
@@ -242,8 +245,13 @@ void SecurityTest::final_message_process_ok(
     info.guid = remote_participant_key;
     EXPECT_CALL(*participant_.getListener(), onParticipantAuthentication(_, info)).Times(1);
 
+    CacheChange_t* kx_change = new CacheChange_t(200);
+    expect_kx_exchange(kx_change);
+
     stateless_reader_->listener_->onNewCacheChangeAdded(stateless_reader_, change);
 
+    volatile_writer_->listener_->onWriterChangeReceivedByAll(volatile_writer_, kx_change);
+
     if (final_message_change == nullptr)
     {
         delete change2;
@@ -254,6 +262,21 @@ void SecurityTest::final_message_process_ok(
     }
 }
 
+void SecurityTest::expect_kx_exchange(
+        CacheChange_t* kx_change)
+{
+    EXPECT_CALL(*volatile_writer_, new_change(_, _, _)).Times(1).WillOnce(
+        DoAll(Invoke([kx_change](const std::function<uint32_t()>& f, ChangeKind_t, InstanceHandle_t)
+        {
+            kx_change->serializedPayload.reserve(f());
+        }),
+        Return(kx_change)));
+    EXPECT_CALL(*volatile_writer_->history_, add_change_mock(kx_change)).Times(1).
+            WillOnce(Return(true));
+    EXPECT_CALL(*volatile_writer_->history_, remove_change_mock(kx_change)).Times(1).
+            WillOnce(Return(true));
+}
+
 void SecurityTest::destroy_manager_and_change(
         CacheChange_t*& change,
         bool was_added)

test/unittest/rtps/security/SecurityTests.hpp

@@ -124,6 +124,9 @@ class SecurityTest : public ::testing::Test
     void final_message_process_ok(
             CacheChange_t** final_message_change = nullptr);
 
+    void expect_kx_exchange(
+            CacheChange_t* kx_change);
+
     void destroy_manager_and_change(
             CacheChange_t*& change,
             bool was_added = true);
@@ -159,7 +162,7 @@ class SecurityTest : public ::testing::Test
     ::testing::NiceMock<RTPSParticipantImpl> participant_;
     ::testing::NiceMock<StatelessWriter>* stateless_writer_;
     ::testing::NiceMock<StatelessReader>* stateless_reader_;
-    ::testing::NiceMock<StatefulWriter>* volatile_writer_;
+    ::testing::StrictMock<StatefulWriter>* volatile_writer_;
     ::testing::NiceMock<StatefulReader>* volatile_reader_;
     PDP pdp_;
     SecurityPluginFactory plugin_factory_;

test/unittest/rtps/security/SecurityValidationRemoteTests.cpp

@@ -158,8 +158,13 @@ TEST_F(SecurityTest, discovered_participant_validation_remote_identity_pending_h
     info.guid = participant_data.m_guid;
     EXPECT_CALL(*participant_.getListener(), onParticipantAuthentication(_, info)).Times(1);
 
+    CacheChange_t* kx_change = new CacheChange_t(500);
+    expect_kx_exchange(kx_change);
+
     ASSERT_TRUE(manager_.discovered_participant(participant_data));
 
+    volatile_writer_->listener_->onWriterChangeReceivedByAll(volatile_writer_, kx_change);
+
     return_handle(remote_identity_handle);
     return_handle(handshake_handle);
 }
@@ -325,8 +330,13 @@ TEST_F(SecurityTest, discovered_participant_validation_remote_identity_pending_h
     info.guid = participant_data.m_guid;
     EXPECT_CALL(*participant_.getListener(), onParticipantAuthentication(_, info)).Times(1);
 
+    CacheChange_t* kx_change = new CacheChange_t(500);
+    expect_kx_exchange(kx_change);
+
     ASSERT_TRUE(manager_.discovered_participant(participant_data));
 
+    volatile_writer_->listener_->onWriterChangeReceivedByAll(volatile_writer_, kx_change);
+
     destroy_manager_and_change(change);
 
     return_handle(remote_identity_handle);