CVE-2023-50257 (C++, published 2024-02-19T20:15Z)
Description: value: eProsima Fast DDS (formerly Fast RTPS) is a C++ implementation of the Data Distribution Service standard of the Object Management Group. Even with the application of SROS2, due to the issue where the data (`p[UD]`) and `guid` values used to disconnect between nodes are not encrypted, a vulnerability has been discovered where a malicious attacker can forcibly disconnect a Subscriber and can deny a Subscriber attempting to connect. Afterwards, if the attacker sends the packet for disconnecting, which is data (`p[UD]`), to the Global Data Space (`239.255.0.1:7400`) using the said Publisher ID, all the Subscribers (Listeners) connected to the Publisher (Talker) will not receive any data and their connection will be disconnected. Moreover, if this disconnection packet is sent continuously, the Subscribers (Listeners) trying to connect will not be able to do so. Since the initial commit of the `SecurityManager.cpp` code (`init`, `on_process_handshake`) on Nov 8, 2016, the Disconnect Vulnerability in RTPS Packets Used by SROS2 has been present prior to versions 2.13.0, 2.12.2, 2.11.3, 2.10.3, and 2.6.7.
Severity: nan (CVSS v3 base score: )
Weakness: CWE-284 \u2014 Improper Access Control
The product does not restrict or incorrectly restricts access to a resource from an unauthorized actor.
Repository: https://github.com/eProsima/Fast-DDS
Files changed:
include/fastdds/rtps/builtin/discovery/participant/PDP.h
include/fastdds/rtps/builtin/discovery/participant/PDPListener.h
include/fastdds/rtps/builtin/discovery/participant/PDPSimple.h
src/cpp/rtps/builtin/BuiltinReader.hpp
src/cpp/rtps/builtin/discovery/endpoint/EDPSimple.cpp
src/cpp/rtps/builtin/discovery/participant/DS/DiscoveryServerPDPEndpoints.hpp
src/cpp/rtps/builtin/discovery/participant/DS/DiscoveryServerPDPEndpointsSecure.hpp
src/cpp/rtps/builtin/discovery/participant/DS/PDPSecurityInitiatorListener.cpp
src/cpp/rtps/builtin/discovery/participant/DS/PDPSecurityInitiatorListener.hpp
src/cpp/rtps/builtin/discovery/participant/PDP.cpp
src/cpp/rtps/builtin/discovery/participant/PDPClient.cpp
src/cpp/rtps/builtin/discovery/participant/PDPEndpoints.hpp
src/cpp/rtps/builtin/discovery/participant/PDPListener.cpp
src/cpp/rtps/builtin/discovery/participant/PDPServer.cpp
src/cpp/rtps/builtin/discovery/participant/PDPSimple.cpp
src/cpp/rtps/builtin/discovery/participant/simple/SimplePDPEndpoints.hpp
src/cpp/rtps/builtin/discovery/participant/simple/SimplePDPEndpointsSecure.hpp
src/cpp/rtps/reader/StatelessReader.cpp
src/cpp/rtps/security/SecurityManager.cpp
test/blackbox/common/BlackboxTestsSecurity.cpp
test/mock/rtps/RTPSParticipantImpl/rtps/participant/RTPSParticipantImpl.h
test/unittest/rtps/discovery/PDPTests.cpp
Commit message: Fix CVE-2023-50257

Signed-off-by: Miguel Company <miguelcompany@eprosima.com>
Co-authored-by: Mario Dominguez <mariodominguez@eprosima.com>

commit f2bcd264798ebd0d45241e57f7d522c68920b26f
Author: Miguel Company <miguelcompany@eprosima.com>
Date:   Thu Dec 14 10:52:32 2023 +0100

    Refs #19770. Improve BuiltinAuthenticationPlugin_second_participant_creation_loop.

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

commit 07380f840422f8ea2a6ae62ec901fe94687f3b3c
Author: Miguel Company <miguelcompany@eprosima.com>
Date:   Wed Dec 13 08:21:34 2023 +0100

    Refs #19770. Fix PDPTests.

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

commit 4c2c21e5f2be63c1f4e0766cc8bc5a59a857b106
Author: Miguel Company <miguelcompany@eprosima.com>
Date:   Wed Dec 13 07:26:51 2023 +0100

    Refs #19770. Adding override keyword.

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

commit d185a85b937a1fd64f86a40b9db2d0b994e39a7f
Author: Miguel Company <miguelcompany@eprosima.com>
Date:   Tue Dec 12 16:05:00 2023 +0100

    Refs #19770. Uncrustify.

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

commit 54a01ab0eff0e326c868d1a32bd44203bf184b4d
Author: Miguel Company <miguelcompany@eprosima.com>
Date:   Tue Dec 12 10:29:42 2023 +0100

    Refs #19770. PDPSecurityInitiatorListener relies on PDPListener::onNewCacheChangeAdded.

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

commit 202bc52a4584025c897e06a40f50d5d7d6be8ce6
Author: Miguel Company <miguelcompany@eprosima.com>
Date:   Tue Dec 12 09:49:20 2023 +0100

    Refs #19770. Only process removals when reader is matched.

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

commit a4a30779c0ae3b410b8cce4aa56f47490995d20d
Author: Miguel Company <miguelcompany@eprosima.com>
Date:   Tue Dec 12 09:46:03 2023 +0100

    Refs #19770. Added PDPSecurityInitiatorListener::process_alive_data.

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

commit 3c7547e0b125ab7156aaf3f72b1a431a9d77672f
Author: Miguel Company <miguelcompany@eprosima.com>
Date:   Tue Dec 12 09:32:55 2023 +0100

    Refs #19770. Refactor with PDPListener::process_alive_data.

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

commit 4571178679af32902a5ba6719807b75eabafc296
Author: Miguel Company <MiguelCompany@eprosima.com>
Date:   Wed Nov 15 12:28:14 2023 +0100

    Refs #19770. StatelessReader does not call update_last_notified for not matched writers.

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

commit 8294ca3b0bae320f066779dfa1d9a8690b16ebfc
Author: Miguel Company <MiguelCompany@eprosima.com>
Date:   Wed Nov 29 13:30:14 2023 +0100

    Refs #19770. Avoid dynamic_cast when participant is not secure.

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

commit 83820e84c6cc3dcaf53e6ffe784ed03b2fac9bfa
Author: Miguel Company <MiguelCompany@eprosima.com>
Date:   Wed Nov 29 13:24:59 2023 +0100

    Refs #19770. Avoid warning when built without security.

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

commit 8ce1216f6af4758f4cb8dc1ea63c770a5ff86e1a
Author: Miguel Company <MiguelCompany@eprosima.com>
Date:   Wed Nov 29 12:55:28 2023 +0100

    Refs #19770. Fixed typo in comments.

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

commit 7bb64eb4a0b899b8055beb8a8b0b279db80e20d6
Author: Miguel Company <MiguelCompany@eprosima.com>
Date:   Wed Nov 29 12:52:07 2023 +0100

    Refs #19770. Use constexpr for topic names.

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

commit eef95273d5464210c50a01cb22cc4a3989233f2f
Author: Miguel Company <MiguelCompany@eprosima.com>
Date:   Wed Nov 29 12:50:56 2023 +0100

    Refs #19770. Send participant dispose messages through both writers.

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

commit c3c831f5f047c4a02a3ddd9944ce3015644f3e27
Author: Miguel Company <MiguelCompany@eprosima.com>
Date:   Tue Nov 28 10:07:17 2023 +0100

    Refs #19770. Move to `AUTHENTICATION_FAILED` when authentication fails.

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

commit 11f98a4d0132d630b76c68ed4157a8624dcffaa5
Author: Miguel Company <MiguelCompany@eprosima.com>
Date:   Tue Nov 28 10:05:47 2023 +0100

    Refs #19770. Simulate DATA(p) reception on non secure `notifyAboveRemoteEndpoints`.

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

commit 5097288e5f4816e653a7c5a4c4839d2a077f3d19
Author: Miguel Company <MiguelCompany@eprosima.com>
Date:   Wed Nov 22 09:36:14 2023 +0100

    Refs #19770. Common factor method `notify_and_maybe_ignore_new_participant`.

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

commit 9364310c5c3cf47b586470b90f9511b0be11fbc4
Author: Miguel Company <MiguelCompany@eprosima.com>
Date:   Tue Nov 21 07:56:17 2023 +0100

    Refs #19770. Improve AllowUnauthenticatedParticipants blackbox tests.

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

commit bb8af40be3c8c138e48acfced5835f2066d864dd
Author: Miguel Company <MiguelCompany@eprosima.com>
Date:   Mon Nov 20 09:53:10 2023 +0100

    Refs #19770. Change builtin endpoints matching workflow.

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

commit 4fc685f75949637ee07ede7a72dd43ed4ab351f3
Author: Miguel Company <MiguelCompany@eprosima.com>
Date:   Fri Nov 17 12:34:40 2023 +0100

    Refs #19770. Chages in announceParticipantState.

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

commit fb26bbccf6400123d2cb0d89d34f8c907e301d1a
Author: Miguel Company <MiguelCompany@eprosima.com>
Date:   Fri Nov 17 12:23:48 2023 +0100

    Refs #19770. Chages in removeRemoteEndpoints.

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

commit 96442d58fa0b16984289bbe16a132769e9cbcd81
Author: Miguel Company <MiguelCompany@eprosima.com>
Date:   Fri Nov 17 12:15:35 2023 +0100

    Refs #19770. Chages in notifyAboveRemoteEndpoints.

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

commit cea35be6c84e633a500bd4565119d1a7ec6066da
Author: Miguel Company <MiguelCompany@eprosima.com>
Date:   Fri Nov 17 11:52:15 2023 +0100

    Refs #19770. Additional work on endpoints creation.

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

commit 02f95cf90b8acdf84c0c6dd357b8a7a26816db93
Author: Miguel Company <MiguelCompany@eprosima.com>
Date:   Fri Nov 17 10:46:48 2023 +0100

    Refs #19770. Methods in PDP to setup builtin endpoints security attributes.

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

commit 1f5c43734dfc6d420103e16132034e7e85ba0c3a
Author: Miguel Company <MiguelCompany@eprosima.com>
Date:   Fri Nov 17 10:38:57 2023 +0100

    Refs #19770. Methods in PDP to create builtin endpoints attributes.

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

commit 49624e88746389b44384f007dc17610ba929731a
Author: Miguel Company <MiguelCompany@eprosima.com>
Date:   Thu Nov 16 16:56:25 2023 +0100

    Refs #19770. Endpoints creation refactor.

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

commit 4ab8d1e1e24819da5d27009ec3e80051329dfbb0
Author: Miguel Company <MiguelCompany@eprosima.com>
Date:   Thu Nov 16 16:28:15 2023 +0100

    Refs #19770. Fix unprotected access to local participant data.

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

commit 57b99075eae290412b4d08798e1df27fc9972e1c
Author: Miguel Company <MiguelCompany@eprosima.com>
Date:   Thu Nov 16 15:38:24 2023 +0100

    Refs #19770. Added SimplePDPEndpointsSecure.

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

commit e67c703db6956e15e187868f90f1f99b25e37627
Author: Miguel Company <MiguelCompany@eprosima.com>
Date:   Thu Nov 16 15:28:38 2023 +0100

    Refs #19770. Listeners moved into BuiltinReader.

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

commit b2fe97e1e0001b24b18d4b145143139a084dc37d
Author: Mario Dominguez <mariodominguez@eprosima.com>
Date:   Wed Nov 8 15:42:34 2023 +0100

    Refs #19770. PDP listener moved into PDPEndpoints.

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

commit 1af9da00af27591be14a1abe1d9928ccff83048d
Author: Miguel Company <MiguelCompany@eprosima.com>
Date:   Thu Nov 16 13:33:37 2023 +0100

    Refs #19770. Test improvements.

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

commit 03eeda5f6b305768cc876056ff6584c153c947d1
Author: Miguel Company <MiguelCompany@eprosima.com>
Date:   Thu Nov 16 13:21:45 2023 +0100

    Refs #19770. Regression test for vulnerability.

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

commit 6cd538753fb8be550f9b3a58c508021d2bdf6ccd
Author: Miguel Company <MiguelCompany@eprosima.com>
Date:   Thu Nov 16 12:30:36 2023 +0100

    Refs #19770. Refactor on BlackboxTestsSecurity.

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

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


Vulnerable code:
    ParticipantProxyData* getLocalParticipantProxyData()
        return participant_proxies_.front();
    //!Listener for the SPDP messages.
    ReaderListener* mp_listener;
class PDPListener: public ReaderListener
    PDPListener(PDP* parent);
    * New added cache
    * @param reader
    * @param change
    */
    bool get_key(CacheChange_t* change);
#endif
    const RTPSParticipantAttributes& pattr = mp_PDP->getRTPSParticipant()->getRTPSParticipantAttributes();

    // Matched writers will depend on total number of participants
    attributes.matched_writers_allocation = pattr.allocation.participants;

    // As participants allocation policy includes the local participant, one has to be substracted
    if (attributes.matched_writers_allocation.initial > 1)
    {
        attributes.matched_writers_allocation.initial--;
    }
    if ((attributes.matched_writers_allocation.maximum > 1) &&
            (attributes.matched_writers_allocation.maximum < std::numeric_limits<size_t>::max()))
    {
        attributes.matched_writers_allocation.maximum--;
    }

    // Locators are copied from the local participant metatraffic locators
    attributes.endpoint.unicastLocatorList.clear();
    for (const Locator_t& loc : mp_PDP->getLocalParticipantProxyData()->metatraffic_locators.unicast)
    {
        attributes.endpoint.unicastLocatorList.push_back(loc);
    }
    attributes.endpoint.multicastLocatorList.clear();
    for (const Locator_t& loc : mp_PDP->getLocalParticipantProxyData()->metatraffic_locators.multicast)
    {
        attributes.endpoint.multicastLocatorList.push_back(loc);
    }
    attributes.endpoint.external_unicast_locators = mp_PDP->builtin_attributes().metatraffic_external_unicast_locators;
    attributes.endpoint.ignore_non_matching_locators = pattr.ignore_non_matching_locators;

    // EDP endpoints are always reliable, transsient local, keyed topics
    attributes.endpoint.reliabilityKind = RELIABLE;
    attributes.endpoint.durabilityKind = TRANSIENT_LOCAL;
    attributes.endpoint.topicKind = WITH_KEY;

    // Built-in EDP readers never expect inline qos
    attributes.expectsInlineQos = false;
    const RTPSParticipantAttributes& pattr = mp_PDP->getRTPSParticipant()->getRTPSParticipantAttributes();

    // Matched readers will depend on total number of participants
    attributes.matched_readers_allocation = pattr.allocation.participants;

    // As participants allocation policy includes the local participant, one has to be substracted
    if (attributes.matched_readers_allocation.initial > 1)
    {
        attributes.matched_readers_allocation.initial--;
    }
    if ((attributes.matched_readers_allocation.maximum > 1) &&
            (attributes.matched_readers_allocation.maximum < std::numeric_limits<size_t>::max()))
    {
        attributes.matched_readers_allocation.maximum--;
    }

    // Locators are copied from the local participant metatraffic locators
    attributes.endpoint.unicastLocatorList.clear();
    for (const Locator_t& loc : mp_PDP->getLocalParticipantProxyData()->metatraffic_locators.unicast)
    {
        attributes.endpoint.unicastLocatorList.push_back(loc);
    }
    attributes.endpoint.multicastLocatorList.clear();
    for (const Locator_t& loc : mp_PDP->getLocalParticipantProxyData()->metatraffic_locators.multicast)
    {
        attributes.endpoint.multicastLocatorList.push_back(loc);
    }
    attributes.endpoint.external_unicast_locators = mp_PDP->builtin_attributes().metatraffic_external_unicast_locators;
    attributes.endpoint.ignore_non_matching_locators = pattr.ignore_non_matching_locators;

    // EDP endpoints are always reliable, transsient local, keyed topics
    attributes.endpoint.reliabilityKind = RELIABLE;
    attributes.endpoint.durabilityKind = TRANSIENT_LOCAL;
    attributes.endpoint.topicKind = WITH_KEY;

    const security::ParticipantSecurityAttributes& part_attr = mp_RTPSParticipant->security_attributes();

    ratt.endpoint.security_attributes().is_submessage_protected = part_attr.is_discovery_protected;
    ratt.endpoint.security_attributes().plugin_endpoint_attributes = PLUGIN_ENDPOINT_SECURITY_ATTRIBUTES_FLAG_IS_VALID;

    watt.endpoint.security_attributes().is_submessage_protected = part_attr.is_discovery_protected;
    watt.endpoint.security_attributes().plugin_endpoint_attributes = PLUGIN_ENDPOINT_SECURITY_ATTRIBUTES_FLAG_IS_VALID;

    if (part_attr.is_discovery_protected)
    {
        security::PluginParticipantSecurityAttributes plugin_part_attr(part_attr.plugin_participant_attributes);

        if (plugin_part_attr.is_discovery_encrypted)
        {
            ratt.endpoint.security_attributes().plugin_endpoint_attributes |=
                    PLUGIN_ENDPOINT_SECURITY_ATTRIBUTES_FLAG_IS_SUBMESSAGE_ENCRYPTED;
            watt.endpoint.security_attributes().plugin_endpoint_attributes |=
                    PLUGIN_ENDPOINT_SECURITY_ATTRIBUTES_FLAG_IS_SUBMESSAGE_ENCRYPTED;
        }
        if (plugin_part_attr.is_discovery_origin_authenticated)
        {
            ratt.endpoint.security_attributes().plugin_endpoint_attributes |=
                    PLUGIN_ENDPOINT_SECURITY_ATTRIBUTES_FLAG_IS_SUBMESSAGE_ORIGIN_AUTHENTICATED;
            watt.endpoint.security_attributes().plugin_endpoint_attributes |=
                    PLUGIN_ENDPOINT_SECURITY_ATTRIBUTES_FLAG_IS_SUBMESSAGE_ORIGIN_AUTHENTICATED;
        }
    }
    //! Listener for stateless_reader
    std::unique_ptr<fastrtps::rtps::ReaderListener> stateless_listener;
    : parent_pdp_(parent)
    , temp_participant_data_(parent->getRTPSParticipant()->getRTPSParticipantAttributes().allocation)
void PDPSecurityInitiatorListener::onNewCacheChangeAdded(
        const CacheChange_t* const change_in)
    CacheChange_t* change = const_cast<CacheChange_t*>(change_in);
    GUID_t writer_guid = change->writerGUID;
    EPROSIMA_LOG_INFO(RTPS_PDP, "SPDP Message received from: " << change_in->writerGUID);

    // Make sure we have an instance handle (i.e GUID)
    if (change->instanceHandle == c_InstanceHandle_Unknown)
        if (!this->get_key(change))
        {
            EPROSIMA_LOG_WARNING(RTPS_PDP, "Problem getting the key of the change, removing");
            parent_pdp_->builtin_endpoints_->remove_from_pdp_reader_history(change);
            return;
        }
    // Take GUID from instance handle
    GUID_t guid;
    iHandle2GUID(guid, change->instanceHandle);

    if (change->kind == ALIVE)
        // Ignore announcement from own RTPSParticipant
        if (guid == parent_pdp_->getRTPSParticipant()->getGuid())
        {
            EPROSIMA_LOG_INFO(RTPS_PDP, "Message from own RTPSParticipant, removing");
            parent_pdp_->builtin_endpoints_->remove_from_pdp_reader_history(change);
            return;
        }

        // Release reader lock to avoid ABBA lock. PDP mutex should always be first.
        // Keep change information on local variables to check consistency later
        SequenceNumber_t seq_num = change->sequenceNumber;
        std::unique_lock<std::recursive_mutex> lock(*parent_pdp_->getMutex());
        reader->getMutex().lock();

        // If change is not consistent, it will be processed on the thread that has overriten it
        if ((ALIVE != change->kind) || (seq_num != change->sequenceNumber) || (writer_guid != change->writerGUID))
        {
            return;
        }

        // Access to temp_participant_data_ is protected by reader lock
        // Load information on temp_participant_data_
        CDRMessage_t msg(change->serializedPayload);
        temp_participant_data_.clear();
        if (temp_participant_data_.readFromCDRMessage(&msg, true, parent_pdp_->getRTPSParticipant()->network_factory(),
                parent_pdp_->getRTPSParticipant()->has_shm_transport(), true))
            // After correctly reading it
            change->instanceHandle = temp_participant_data_.m_key;
            guid = temp_participant_data_.m_guid;

            // Filter locators
            const auto& pattr = parent_pdp_->getRTPSParticipant()->getAttributes();
            fastdds::rtps::ExternalLocatorsProcessor::filter_remote_locators(temp_participant_data_,
                    pattr.builtin.metatraffic_external_unicast_locators, pattr.default_external_unicast_locators,
                    pattr.ignore_non_matching_locators);

            // Check if participant already exists (updated info)
            ParticipantProxyData* pdata = nullptr;
            for (ParticipantProxyData* it : parent_pdp_->participant_proxies_)
            {
                if (guid == it->m_guid)
                {
                    pdata = it;
                    break;
                }
            }

            if (pdata == nullptr)
            {
                // Create a new one when not found

                reader->getMutex().unlock();
                lock.unlock();

                //! notify security manager in order to start handshake
                bool ret = parent_pdp_->getRTPSParticipant()->security_manager().discovered_participant(
                    temp_participant_data_);

                //! Reply to the remote participant
                if (ret)
                {
                    response_cb_(temp_participant_data_);
                }

                // Take again the reader lock
                reader->getMutex().lock();

            } //! Do nothing if already discovered

    } //! Do nothing on participant removal

    //Remove change form history.
    parent_pdp_->builtin_endpoints_->remove_from_pdp_reader_history(change);
}
bool PDPSecurityInitiatorListener::get_key(
        CacheChange_t* change)
{
    return ParameterList::readInstanceHandleFromCDRMsg(change, fastdds::dds::PID_PARTICIPANT_GUID);
#include <fastdds/rtps/reader/ReaderListener.h>
class PDPSecurityInitiatorListener : public ReaderListener
    /**
     * New added cache
     * @param reader
     * @param change
     */
    void onNewCacheChangeAdded(
            RTPSReader* reader,
            const CacheChange_t* const change) override;

    /**
     * Get the key of a CacheChange_t
     * @param change Pointer to the CacheChange_t
     * @return True on success
     */
    bool get_key(
            CacheChange_t* change);

    //!Pointer to the associated mp_SPDP;
    PDP* parent_pdp_;

    /**
     * @brief Temporary data to avoid reallocations.
     *
     * @remarks This should be always accessed with the pdp_reader lock taken
     */
    ParticipantProxyData temp_participant_data_;


    , mp_listener(nullptr)
    delete mp_listener;
    mp_listener = nullptr;

            ParticipantProxyData proxy_data_copy(*getLocalParticipantProxyData());
                            NOT_ALIVE_DISPOSED_UNREGISTERED, getLocalParticipantProxyData()->m_key);
    endpoints.stateless_listener.reset(new PDPSecurityInitiatorListener(this));
            endpoints.stateless_listener.get(), c_EntityId_SPDPReader, true, false))

        // Enable unknown clients to reach this reader
        reader->enableMessagesFromUnkownWriters(true);


        endpoints.stateless_listener.reset();
    mp_listener = new PDPListener(this);
    if (mp_RTPSParticipant->createReader(&reader, ratt, endpoints.reader.history_.get(), mp_listener,
        delete mp_listener;
        mp_listener = nullptr;
        eprosima::shared_lock<eprosima::shared_mutex> disc_lock(mp_builtin->getDiscoveryMutex());

        // Verify if this participant is a server
        for (auto& svr : mp_builtin->m_DiscoveryServers)
            if (data_matches_with_prefix(svr.guidPrefix, *pdata))
                std::unique_lock<std::recursive_mutex> lock(*getMutex());
                svr.proxy = pdata;
    }
    if (mp_RTPSParticipant->security_manager().discovered_participant(*pdata))
    {
        perform_builtin_endpoints_matching(*pdata);
    EPROSIMA_LOG_INFO(RTPS_PDP, "SPDP Message received from: " << change_in->writerGUID);
            auto status = (pdata == nullptr) ? ParticipantDiscoveryInfo::DISCOVERED_PARTICIPANT :
                    ParticipantDiscoveryInfo::CHANGED_QOS_PARTICIPANT;

            if (pdata == nullptr)
            {
                // Create a new one when not found
                pdata = parent_pdp_->createParticipantProxyData(temp_participant_data_, writer_guid);

                reader->getMutex().unlock();
                lock.unlock();

                if (pdata != nullptr)
                {
                    EPROSIMA_LOG_INFO(RTPS_PDP_DISCOVERY, "New participant "
                            << pdata->m_guid << " at "
                            << "MTTLoc: " << pdata->metatraffic_locators
                            << " DefLoc:" << pdata->default_locators);

                    RTPSParticipantListener* listener = parent_pdp_->getRTPSParticipant()->getListener();
                    if (listener != nullptr)
                    {
                        bool should_be_ignored = false;
                        {
                            std::lock_guard<std::mutex> cb_lock(parent_pdp_->callback_mtx_);
                            ParticipantDiscoveryInfo info(*pdata);
                            info.status = status;


                            listener->onParticipantDiscovery(
                                parent_pdp_->getRTPSParticipant()->getUserRTPSParticipant(),
                                std::move(info),
                                should_be_ignored);
                        }
                        if (should_be_ignored)
                        {
                            parent_pdp_->getRTPSParticipant()->ignore_participant(guid.guidPrefix);
                        }

                    }

                    // Assigning remote endpoints implies sending a DATA(p) to all matched and fixed readers, since
                    // StatelessWriter::matched_reader_add marks the entire history as unsent if the added reader's
                    // durability is bigger or equal to TRANSIENT_LOCAL_DURABILITY_QOS (TRANSIENT_LOCAL or TRANSIENT),
                    // which is the case of ENTITYID_BUILTIN_SDP_PARTICIPANT_READER (TRANSIENT_LOCAL). If a remote
                    // participant is discovered before creating the first DATA(p) change (which happens at the end of
                    // BuiltinProtocols::initBuiltinProtocols), then StatelessWriter::matched_reader_add ends up marking
                    // no changes as unsent (since the history is empty), which is OK because this can only happen if a
                    // participant is discovered in the middle of BuiltinProtocols::initBuiltinProtocols, which will
                    // create the first DATA(p) upon finishing, thus triggering the sent to all fixed and matched
                    // readers anyways.
                    parent_pdp_->assignRemoteEndpoints(pdata);
                }
            }
            else
            {
                pdata->updateData(temp_participant_data_);
                pdata->isAlive = true;
                reader->getMutex().unlock();

                EPROSIMA_LOG_INFO(RTPS_PDP_DISCOVERY, "Update participant "
                        << pdata->m_guid << " at "
                        << "MTTLoc: " << pdata->metatraffic_locators
                        << " DefLoc:" << pdata->default_locators);

                if (parent_pdp_->updateInfoMatchesEDP())
                {
                    parent_pdp_->mp_EDP->assignRemoteEndpoints(*pdata, true);
                }

                lock.unlock();

                RTPSParticipantListener* listener = parent_pdp_->getRTPSParticipant()->getListener();
                if (listener != nullptr)
                {
                    bool should_be_ignored = false;

                    {
                        std::lock_guard<std::mutex> cb_lock(parent_pdp_->callback_mtx_);
                        ParticipantDiscoveryInfo info(*pdata);
                        info.status = status;

                        listener->onParticipantDiscovery(
                            parent_pdp_->getRTPSParticipant()->getUserRTPSParticipant(),
                            std::move(info),
                            should_be_ignored);
                    }
                    if (should_be_ignored)
                    {
                        parent_pdp_->getRTPSParticipant()->ignore_participant(temp_participant_data_.m_guid.guidPrefix);
                    }
                }
            }

#ifdef FASTDDS_STATISTICS
            //! Addition or update of a participant proxy should trigger
            //! a connections update on the local participant connection list
            if (nullptr != parent_pdp_->getRTPSParticipant()->get_connections_observer())
            {
                parent_pdp_->getRTPSParticipant()->get_connections_observer()->on_local_entity_connections_change(
                    parent_pdp_->getRTPSParticipant()->getGuid());
            }
#endif //FASTDDS_STATISTICS

            // Take again the reader lock
            reader->getMutex().lock();
    else
    endpoints.stateless_listener.reset(new PDPSecurityInitiatorListener(this,
            endpoints.stateless_listener.get(), c_EntityId_SPDPReader, true, false))

        // Enable unknown clients to reach this reader
        reader->enableMessagesFromUnkownWriters(true);


        endpoints.stateless_listener.reset();
    mp_listener = new PDPServerListener(this);
    if (mp_RTPSParticipant->createReader(&reader, ratt, endpoints.reader.history_.get(), mp_listener,
            reader_entity, true, false))
        delete mp_listener;
        mp_listener = nullptr;
                mp_listener->onNewCacheChangeAdded(endpoints->reader.reader_, change_aux);
using namespace eprosima::fastrtps;

        auto endpoints = static_cast<fastdds::rtps::SimplePDPEndpoints*>(builtin_endpoints_.get());
        StatelessWriter& writer = *(endpoints->writer.writer_);
        WriterHistory& history = *(endpoints->writer.history_);
            writer.unsent_changes_reset();
    auto endpoints = new fastdds::rtps::SimplePDPEndpoints();
    builtin_endpoints_.reset(endpoints);
    //SPDP BUILTIN RTPSParticipant READER
    hatt.payloadMaxSize = builtin_att.readerPayloadSize;
    hatt.memoryPolicy = builtin_att.readerHistoryMemoryPolicy;
    hatt.initialReservedCaches = 25;
    if (allocation.participants.initial > 0)
    {
        hatt.initialReservedCaches = (int32_t)allocation.participants.initial;
    }
    if (allocation.participants.maximum < std::numeric_limits<size_t>::max())
    {
        hatt.maximumReservedCaches = (int32_t)allocation.participants.maximum;
    }
    endpoints->reader.payload_pool_ = TopicPayloadPoolRegistry::get("DCPSParticipant", reader_pool_cfg);
    endpoints->reader.payload_pool_->reserve_history(reader_pool_cfg, true);

    endpoints->reader.history_.reset(new ReaderHistory(hatt));

    ReaderAttributes ratt;
    ratt.endpoint.multicastLocatorList = mp_builtin->m_metatrafficMulticastLocatorList;
    ratt.endpoint.unicastLocatorList = mp_builtin->m_metatrafficUnicastLocatorList;
    ratt.endpoint.external_unicast_locators = mp_builtin->m_att.metatraffic_external_unicast_locators;
    ratt.endpoint.ignore_non_matching_locators = pattr.ignore_non_matching_locators;
    ratt.endpoint.topicKind = WITH_KEY;
    ratt.endpoint.durabilityKind = TRANSIENT_LOCAL;
    ratt.matched_writers_allocation = allocation.participants;
    mp_listener = new PDPListener(this);
    RTPSReader* reader = nullptr;
    if (mp_RTPSParticipant->createReader(&reader, ratt,
            endpoints->reader.payload_pool_, endpoints->reader.history_.get(),
            mp_listener, c_EntityId_SPDPReader, true, false))
    {
        endpoints->reader.reader_ = dynamic_cast<StatelessReader*>(reader);
        mp_RTPSParticipant->set_endpoint_rtps_protection_supports(reader, false);
        EPROSIMA_LOG_ERROR(RTPS_PDP, "SimplePDP Reader creation failed");
        delete mp_listener;
        mp_listener = nullptr;
        endpoints->reader.release();
    //SPDP BUILTIN RTPSParticipant WRITER
    hatt.payloadMaxSize = mp_builtin->m_att.writerPayloadSize;
    hatt.initialReservedCaches = 1;
    hatt.maximumReservedCaches = 1;
    hatt.memoryPolicy = mp_builtin->m_att.writerHistoryMemoryPolicy;
    endpoints->writer.payload_pool_ = TopicPayloadPoolRegistry::get("DCPSParticipant", writer_pool_cfg);
    endpoints->writer.payload_pool_->reserve_history(writer_pool_cfg, false);

    endpoints->writer.history_.reset(new WriterHistory(hatt));
    WriterAttributes watt;
    watt.endpoint.external_unicast_locators = mp_builtin->m_att.metatraffic_external_unicast_locators;
    watt.endpoint.ignore_non_matching_locators = pattr.ignore_non_matching_locators;
    watt.endpoint.endpointKind = WRITER;
    watt.endpoint.durabilityKind = TRANSIENT_LOCAL;
    watt.endpoint.topicKind = WITH_KEY;
    watt.matched_readers_allocation = allocation.participants;
    RTPSWriter* wout = nullptr;
    if (mp_RTPSParticipant->createWriter(&wout, watt, endpoints->writer.payload_pool_, endpoints->writer.history_.get(),
            nullptr,
            c_EntityId_SPDPWriter, true))
        endpoints->writer.writer_ = dynamic_cast<StatelessWriter*>(wout);
        mp_RTPSParticipant->set_endpoint_rtps_protection_supports(wout, false);
        if (endpoints->writer.writer_ != nullptr)
            const NetworkFactory& network = mp_RTPSParticipant->network_factory();
            LocatorList_t fixed_locators;
            for (const Locator_t& loc : mp_builtin->m_initialPeersList)
                if (network.is_locator_remote_or_allowed(loc))
                {
                    // Add initial peers locator without transformation as we don't know whether the
                    // remote transport will allow localhost
                    fixed_locators.push_back(loc);

                    /**
                     * TCP special case:
                     *
                     * In TCP, it is not possible to open a socket with 'any' (0.0.0.0) address as it's done
                     * in UDP, so when the TCP transports receive a locator with 'any', they open an input
                     * channel for the specified port in each of the machine interfaces (with the exception
                     * of localhost). In fact, a participant with a TCP transport will only listen on localhost
                     * if localhost is the address of any of the initial peers.
                     *
                     * However, when the TCP enabled participant does not have a whitelist (or localhost is in
                     * it), it allows for transformation of its locators to localhost for performance optimizations.
                     * In this case, the remote TCP participant it will send data using a socket in localhost,
                     * and for that the participant with the initial peers list needs to be listening there
                     * to receive it.
                     *
                     * That means:
                     *   1. Checking that the initial peer is not already localhost
                     *   2. Checking that the initial peer locator is of TCP kind
                     *   3. Checking that the network configuration allows for localhost locators
                     */
                    Locator_t local_locator;
                    network.transform_remote_locator(loc, local_locator,
                            DISC_NETWORK_CONFIGURATION_LISTENING_LOCALHOST_ALL);
                    if (loc != local_locator
                            && (loc.kind == LOCATOR_KIND_TCPv4 || loc.kind == LOCATOR_KIND_TCPv6)
                            && network.is_locator_allowed(local_locator))
                    {
                        fixed_locators.push_back(local_locator);
                    }
                }
                else
                    EPROSIMA_LOG_WARNING(RTPS_PDP, "Ignoring initial peers locator " << loc << " : not allowed.");
            endpoints->writer.writer_->set_fixed_locators(fixed_locators);
        EPROSIMA_LOG_ERROR(RTPS_PDP, "SimplePDP Writer creation failed");
        endpoints->writer.release();
    EPROSIMA_LOG_INFO(RTPS_PDP, "SPDP Endpoints creation finished");
void PDPSimple::assignRemoteEndpoints(
        ParticipantProxyData* pdata)
    EPROSIMA_LOG_INFO(RTPS_PDP, "For RTPSParticipant: " << pdata->m_guid.guidPrefix);
    auto endpoints = static_cast<fastdds::rtps::SimplePDPEndpoints*>(builtin_endpoints_.get());
    const NetworkFactory& network = mp_RTPSParticipant->network_factory();
    uint32_t endp = pdata->m_availableBuiltinEndpoints;
    uint32_t auxendp = endp;
    bool use_multicast_locators = !mp_RTPSParticipant->getAttributes().builtin.avoid_builtin_multicast ||
            pdata->metatraffic_locators.unicast.empty();
    auxendp &= DISC_BUILTIN_ENDPOINT_PARTICIPANT_ANNOUNCER;
    if (auxendp != 0)
    {
        auto temp_writer_data = get_temporary_writer_proxies_pool().get();
        temp_writer_data->clear();
        temp_writer_data->guid().guidPrefix = pdata->m_guid.guidPrefix;
        temp_writer_data->guid().entityId = c_EntityId_SPDPWriter;
        temp_writer_data->persistence_guid(pdata->get_persistence_guid());
        temp_writer_data->set_persistence_entity_id(c_EntityId_SPDPWriter);
        temp_writer_data->set_remote_locators(pdata->metatraffic_locators, network, use_multicast_locators);
        temp_writer_data->m_qos.m_reliability.kind = RELIABLE_RELIABILITY_QOS;
        temp_writer_data->m_qos.m_durability.kind = TRANSIENT_LOCAL_DURABILITY_QOS;
        endpoints->reader.reader_->matched_writer_add(*temp_writer_data);
    auxendp = endp;
    auxendp &= DISC_BUILTIN_ENDPOINT_PARTICIPANT_DETECTOR;
    if (auxendp != 0)
        auto temp_reader_data = get_temporary_reader_proxies_pool().get();
        temp_reader_data->clear();
        temp_reader_data->m_expectsInlineQos = false;
        temp_reader_data->guid().guidPrefix = pdata->m_guid.guidPrefix;
        temp_reader_data->guid().entityId = c_EntityId_SPDPReader;
        temp_reader_data->set_remote_locators(pdata->metatraffic_locators, network, use_multicast_locators);
        temp_reader_data->m_qos.m_reliability.kind = BEST_EFFORT_RELIABILITY_QOS;
        temp_reader_data->m_qos.m_durability.kind = TRANSIENT_LOCAL_DURABILITY_QOS;
        endpoints->writer.writer_->matched_reader_add(*temp_reader_data);
        StatelessWriter* pW = endpoints->writer.writer_;
        if (pW != nullptr)
            pW->unsent_changes_reset();
            EPROSIMA_LOG_ERROR(RTPS_PDP, "Using PDPSimple protocol with a reliable writer");

#if HAVE_SECURITY
    // Validate remote participant
    mp_RTPSParticipant->security_manager().discovered_participant(*pdata);
#else
    //Inform EDP of new RTPSParticipant data:
    notifyAboveRemoteEndpoints(*pdata, true);
#endif // if HAVE_SECURITY
    auto endpoints = static_cast<fastdds::rtps::SimplePDPEndpoints*>(builtin_endpoints_.get());
    uint32_t endp = pdata->m_availableBuiltinEndpoints;
    uint32_t auxendp = endp;
    auxendp &= DISC_BUILTIN_ENDPOINT_PARTICIPANT_ANNOUNCER;
    if (auxendp != 0)
        GUID_t writer_guid(pdata->m_guid.guidPrefix, c_EntityId_SPDPWriter);
        endpoints->reader.reader_->matched_writer_remove(writer_guid);
    auxendp = endp;
    auxendp &= DISC_BUILTIN_ENDPOINT_PARTICIPANT_DETECTOR;
    if (auxendp != 0)
        GUID_t reader_guid(pdata->m_guid.guidPrefix, c_EntityId_SPDPReader);
        endpoints->writer.writer_->matched_reader_remove(reader_guid);
    //Inform EDP of new RTPSParticipant data:
        mp_EDP->assignRemoteEndpoints(pdata, (notify_secure_endpoints ? true : false));
        mp_builtin->mp_WLP->assignRemoteEndpoints(pdata, (notify_secure_endpoints ? true : false));
        // Update Ownership strength.
        if (EXCLUSIVE_OWNERSHIP_QOS == m_att.ownershipKind)
            auto writer = std::find_if(matched_writers_.begin(), matched_writers_.end(),
            SequenceNumber_t previous_seq = update_last_notified(change->writerGUID, change->sequenceNumber);
                SecurityException exception;
                remote_participant_info->event_->cancel_timer();
                on_validation_failed(dp_it->second->participant_data(), exception);
#include <gtest/gtest.h>
#include <fstream>
#include <map>
#include <algorithm>

#include <fastrtps/xmlparser/XMLProfileManager.h>
#include <rtps/transport/test_UDPv4Transport.h>
#include <fastdds/dds/log/Log.hpp>
TEST_P(Security, BuiltinAuthenticationPlugin_second_participant_creation_loop)
        // Wait for the new participant to be discovered by the main one
        EXPECT_TRUE(main_participant.wait_participant_discovery());
    //! Wait enough time for the PKI requests to time out and give validation_failed (~15secs)
    writer.wait_discovery(std::chrono::seconds(20));
    ASSERT_TRUE(writer.get_matched());
    //! Wait enough time for the PKI requests to time out and give validation_failed (~15secs)
    writer.wait_discovery(std::chrono::seconds(20));
    ASSERT_FALSE(writer.get_matched());
    struct FakeMsg
    {
        asio::io_service service;
        asio::ip::udp::socket socket;

        FakeMsg()
            : service()
            , socket(service)
        {
            socket.open(asio::ip::udp::v4());
        }

        void send(
                const CDRMessage_t& msg,
                const Locator_t& destination)
        {
            std::string addr = IPLocator::toIPv4string(destination);
            unsigned short port = static_cast<unsigned short>(destination.port);
            auto remote = asio::ip::udp::endpoint(asio::ip::address::from_string(addr), port);
            asio::error_code ec;

            socket.send_to(asio::buffer(msg.buffer, msg.length), remote, 0, ec);
        }

    };
    FakeMsg fake_msg;
        fake_msg.send(msg, reader_locator);

Fixed code:
    ParticipantProxyData* getLocalParticipantProxyData() const
        return participant_proxies_.empty() ? nullptr : participant_proxies_.front();
    ReaderAttributes create_builtin_reader_attributes() const;

    WriterAttributes create_builtin_writer_attributes() const;

    void add_builtin_security_attributes(
            ReaderAttributes& ratt,
            WriterAttributes& watt) const;

    void notify_and_maybe_ignore_new_participant(
            ParticipantProxyData* pdata,
            bool& should_be_ignored);

class PDPListener : public ReaderListener

    PDPListener(
            PDP* parent);
     * New added cache
     * @param reader
     * @param change
     */
    /**
     * Process an incoming DATA(p).
     * This method is called from PDPListener::onNewCacheChangeAdded() when an alive DATA(p) is received.
     * Both the reader lock and the PDP lock are held when this method is called.
     *
     * @param  old_data     Pointer to the ParticipantProxyData currently stored in the database. May be nullptr, for a
     *                      recently discovered participant.
     * @param  new_data     ParticipantProxyData from the DATA(p) message.
     * @param  writer_guid  GUID of the writer that sent the DATA(p) message.
     * @param  reader       RTPSReader that received the DATA(p) message.
     * @param  lock         Lock on the PDP database. Passed so it can be released before invoking callbacks.
     */
    virtual void process_alive_data(
            ParticipantProxyData* old_data,
            ParticipantProxyData& new_data,
            GUID_t& writer_guid,
            RTPSReader* reader,
            std::unique_lock<std::recursive_mutex>& lock);

    bool get_key(
            CacheChange_t* change);
#endif // ifndef DOXYGEN_SHOULD_SKIP_THIS_PUBLIC
    bool create_dcps_participant_endpoints();

    void match_pdp_remote_endpoints(
            const ParticipantProxyData& pdata,
            bool notify_secure_endpoints);

    void assign_low_level_remote_endpoints(
            const ParticipantProxyData& pdata,
            bool notify_secure_endpoints);

#if HAVE_SECURITY
    bool create_dcps_participant_secure_endpoints();

    bool pairing_remote_writer_with_local_reader_after_security(
            const GUID_t& local_reader,
            const WriterProxyData& remote_writer_data) override;

    bool pairing_remote_reader_with_local_writer_after_security(
            const GUID_t& local_reader,
            const ReaderProxyData& remote_reader_data) override;
#endif // HAVE_SECURITY

#include <fastdds/rtps/reader/ReaderListener.h>

        listener_.reset();
    //! Listener for the builtin RTPS reader
    std::unique_ptr<fastrtps::rtps::ReaderListener> listener_;
    attributes = mp_PDP->create_builtin_reader_attributes();
    attributes = mp_PDP->create_builtin_writer_attributes();
    mp_PDP->add_builtin_security_attributes(ratt, watt);
    const std::unique_ptr<fastrtps::rtps::ReaderListener>& main_listener() const override
    {
        return reader.listener_;
    }

#include <fastdds/rtps/builtin/discovery/participant/PDPListener.h>
    : PDPListener(parent)
void PDPSecurityInitiatorListener::process_alive_data(
        ParticipantProxyData* old_data,
        ParticipantProxyData& new_data,
        GUID_t& writer_guid,
        std::unique_lock<std::recursive_mutex>& lock)
    if (reader->matched_writer_is_matched(writer_guid))
        // Act as the standard PDPListener when the writer is matched.
        // This will be the case for unauthenticated participants when
        // allowed_unathenticated_participants is true
        PDPListener::process_alive_data(old_data, new_data, writer_guid, reader, lock);
        return;
    if (old_data == nullptr)
        lock.unlock();
        //! notify security manager in order to start handshake
        bool ret = parent_pdp_->getRTPSParticipant()->security_manager().discovered_participant(new_data);
        //! Reply to the remote participant
        if (ret)
            response_cb_(temp_participant_data_);
        // Take again the reader lock
        reader->getMutex().lock();
    }
#include <fastdds/rtps/builtin/discovery/participant/PDPListener.h>
class PDPSecurityInitiatorListener : public PDPListener
    void process_alive_data(
            ParticipantProxyData* old_data,
            ParticipantProxyData& new_data,
            GUID_t& writer_guid,
            RTPSReader* reader,
            std::unique_lock<std::recursive_mutex>& lock) override;
            ParticipantProxyData* local_participant_data = getLocalParticipantProxyData();
            InstanceHandle_t key = local_participant_data->m_key;
            ParticipantProxyData proxy_data_copy(*local_participant_data);
                            NOT_ALIVE_DISPOSED_UNREGISTERED, key);
void PDP::notify_and_maybe_ignore_new_participant(
        ParticipantProxyData* pdata,
        bool& should_be_ignored)
{
    should_be_ignored = false;

    EPROSIMA_LOG_INFO(RTPS_PDP_DISCOVERY, "New participant "
            << pdata->m_guid << " at "
            << "MTTLoc: " << pdata->metatraffic_locators
            << " DefLoc:" << pdata->default_locators);

    RTPSParticipantListener* listener = getRTPSParticipant()->getListener();
    if (listener != nullptr)
    {
        {
            std::lock_guard<std::mutex> cb_lock(callback_mtx_);
            ParticipantDiscoveryInfo info(*pdata);
            info.status = ParticipantDiscoveryInfo::DISCOVERED_PARTICIPANT;


            listener->onParticipantDiscovery(
                getRTPSParticipant()->getUserRTPSParticipant(),
                std::move(info),
                should_be_ignored);
        }

        if (should_be_ignored)
        {
            getRTPSParticipant()->ignore_participant(pdata->m_guid.guidPrefix);
        }
    }
}

static void set_builtin_matched_allocation(
        ResourceLimitedContainerConfig& allocation,
        const RTPSParticipantAttributes& pattr)
{
    // Matched endpoints will depend on total number of participants
    allocation = pattr.allocation.participants;

    // As participants allocation policy includes the local participant, one has to be substracted
    if (allocation.initial > 1)
    {
        allocation.initial--;
    }
    if ((allocation.maximum > 1) &&
            (allocation.maximum < std::numeric_limits<size_t>::max()))
    {
        allocation.maximum--;
    }
}

static void set_builtin_endpoint_locators(
        EndpointAttributes& endpoint,
        const PDP* pdp,
        const BuiltinProtocols* builtin)
{
    const RTPSParticipantAttributes& pattr = pdp->getRTPSParticipant()->getRTPSParticipantAttributes();

    auto part_data = pdp->getLocalParticipantProxyData();
    if (nullptr == part_data)
    {
        // Local participant data has not yet been created.
        // This means we are creating the PDP endpoints, so we copy the locators from mp_builtin
        endpoint.multicastLocatorList = builtin->m_metatrafficMulticastLocatorList;
        endpoint.unicastLocatorList = builtin->m_metatrafficUnicastLocatorList;
    }
    else
    {
        // Locators are copied from the local participant metatraffic locators
        endpoint.unicastLocatorList.clear();
        for (const Locator_t& loc : part_data->metatraffic_locators.unicast)
        {
            endpoint.unicastLocatorList.push_back(loc);
        }
        endpoint.multicastLocatorList.clear();
        for (const Locator_t& loc : part_data->metatraffic_locators.multicast)
        {
            endpoint.multicastLocatorList.push_back(loc);
        }
    }

    // External locators are always taken from the same place
    endpoint.external_unicast_locators = pdp->builtin_attributes().metatraffic_external_unicast_locators;
    endpoint.ignore_non_matching_locators = pattr.ignore_non_matching_locators;
}

ReaderAttributes PDP::create_builtin_reader_attributes() const
{
    ReaderAttributes attributes;

    const RTPSParticipantAttributes& pattr = getRTPSParticipant()->getRTPSParticipantAttributes();
    set_builtin_matched_allocation(attributes.matched_writers_allocation, pattr);
    set_builtin_endpoint_locators(attributes.endpoint, this, mp_builtin);

    // Builtin endpoints are always reliable, transient local, keyed topics
    attributes.endpoint.reliabilityKind = RELIABLE;
    attributes.endpoint.durabilityKind = TRANSIENT_LOCAL;
    attributes.endpoint.topicKind = WITH_KEY;

    // Built-in readers never expect inline qos
    attributes.expectsInlineQos = false;

    return attributes;
}

WriterAttributes PDP::create_builtin_writer_attributes() const
{
    WriterAttributes attributes;

    const RTPSParticipantAttributes& pattr = getRTPSParticipant()->getRTPSParticipantAttributes();
    set_builtin_matched_allocation(attributes.matched_readers_allocation, pattr);
    set_builtin_endpoint_locators(attributes.endpoint, this, mp_builtin);

    // Builtin endpoints are always reliable, transient local, keyed topics
    attributes.endpoint.reliabilityKind = RELIABLE;
    attributes.endpoint.durabilityKind = TRANSIENT_LOCAL;
    attributes.endpoint.topicKind = WITH_KEY;

    return attributes;
}

#if HAVE_SECURITY
void PDP::add_builtin_security_attributes(
        ReaderAttributes& ratt,
        WriterAttributes& watt) const
{
    const security::ParticipantSecurityAttributes& part_attr = mp_RTPSParticipant->security_attributes();

    ratt.endpoint.security_attributes().is_submessage_protected = part_attr.is_discovery_protected;
    ratt.endpoint.security_attributes().plugin_endpoint_attributes = PLUGIN_ENDPOINT_SECURITY_ATTRIBUTES_FLAG_IS_VALID;

    watt.endpoint.security_attributes().is_submessage_protected = part_attr.is_discovery_protected;
    watt.endpoint.security_attributes().plugin_endpoint_attributes = PLUGIN_ENDPOINT_SECURITY_ATTRIBUTES_FLAG_IS_VALID;

    if (part_attr.is_discovery_protected)
    {
        security::PluginParticipantSecurityAttributes plugin_part_attr(part_attr.plugin_participant_attributes);

        if (plugin_part_attr.is_discovery_encrypted)
        {
            ratt.endpoint.security_attributes().plugin_endpoint_attributes |=
                    PLUGIN_ENDPOINT_SECURITY_ATTRIBUTES_FLAG_IS_SUBMESSAGE_ENCRYPTED;
            watt.endpoint.security_attributes().plugin_endpoint_attributes |=
                    PLUGIN_ENDPOINT_SECURITY_ATTRIBUTES_FLAG_IS_SUBMESSAGE_ENCRYPTED;
        }
        if (plugin_part_attr.is_discovery_origin_authenticated)
        {
            ratt.endpoint.security_attributes().plugin_endpoint_attributes |=
                    PLUGIN_ENDPOINT_SECURITY_ATTRIBUTES_FLAG_IS_SUBMESSAGE_ORIGIN_AUTHENTICATED;
            watt.endpoint.security_attributes().plugin_endpoint_attributes |=
                    PLUGIN_ENDPOINT_SECURITY_ATTRIBUTES_FLAG_IS_SUBMESSAGE_ORIGIN_AUTHENTICATED;
        }
    }
}

#endif // HAVE_SECURITY

    endpoints.stateless_reader.listener_.reset(new PDPSecurityInitiatorListener(this));
            endpoints.stateless_reader.listener_.get(), c_EntityId_SPDPReader, true, false))
    endpoints.reader.listener_.reset(new PDPListener(this));
    if (mp_RTPSParticipant->createReader(&reader, ratt, endpoints.reader.history_.get(),
            endpoints.reader.listener_.get(),
    bool ignored = false;
    notify_and_maybe_ignore_new_participant(pdata, ignored);
    if (!ignored)
            eprosima::shared_lock<eprosima::shared_mutex> disc_lock(mp_builtin->getDiscoveryMutex());

            // Verify if this participant is a server
            for (auto& svr : mp_builtin->m_DiscoveryServers)
                if (data_matches_with_prefix(svr.guidPrefix, *pdata))
                {
                    std::unique_lock<std::recursive_mutex> lock(*getMutex());
                    svr.proxy = pdata;
                }
        if (mp_RTPSParticipant->security_manager().discovered_participant(*pdata))
        {
            perform_builtin_endpoints_matching(*pdata);
        }
    virtual const std::unique_ptr<fastrtps::rtps::ReaderListener>& main_listener() const = 0;

    EPROSIMA_LOG_INFO(RTPS_PDP, "SPDP Message received from: " << writer_guid);
            process_alive_data(pdata, temp_participant_data_, writer_guid, reader, lock);
    else if (reader->matched_writer_is_matched(writer_guid))
void PDPListener::process_alive_data(
        ParticipantProxyData* old_data,
        ParticipantProxyData& new_data,
        GUID_t& writer_guid,
        RTPSReader* reader,
        std::unique_lock<std::recursive_mutex>& lock)
{
    GUID_t participant_guid = new_data.m_guid;

    if (old_data == nullptr)
    {
        // Create a new one when not found
        old_data = parent_pdp_->createParticipantProxyData(new_data, writer_guid);

        reader->getMutex().unlock();
        lock.unlock();

        if (old_data != nullptr)
        {
            // Assigning remote endpoints implies sending a DATA(p) to all matched and fixed readers, since
            // StatelessWriter::matched_reader_add marks the entire history as unsent if the added reader's
            // durability is bigger or equal to TRANSIENT_LOCAL_DURABILITY_QOS (TRANSIENT_LOCAL or TRANSIENT),
            // which is the case of ENTITYID_BUILTIN_SDP_PARTICIPANT_READER (TRANSIENT_LOCAL). If a remote
            // participant is discovered before creating the first DATA(p) change (which happens at the end of
            // BuiltinProtocols::initBuiltinProtocols), then StatelessWriter::matched_reader_add ends up marking
            // no changes as unsent (since the history is empty), which is OK because this can only happen if a
            // participant is discovered in the middle of BuiltinProtocols::initBuiltinProtocols, which will
            // create the first DATA(p) upon finishing, thus triggering the sent to all fixed and matched
            // readers anyways.
            parent_pdp_->assignRemoteEndpoints(old_data);
        }
    }
    else
    {
        old_data->updateData(new_data);
        old_data->isAlive = true;
        reader->getMutex().unlock();

        EPROSIMA_LOG_INFO(RTPS_PDP_DISCOVERY, "Update participant "
                << old_data->m_guid << " at "
                << "MTTLoc: " << old_data->metatraffic_locators
                << " DefLoc:" << old_data->default_locators);

        if (parent_pdp_->updateInfoMatchesEDP())
        {
            parent_pdp_->mp_EDP->assignRemoteEndpoints(*old_data, true);
        }

        lock.unlock();

        RTPSParticipantListener* listener = parent_pdp_->getRTPSParticipant()->getListener();
        if (listener != nullptr)
        {
            bool should_be_ignored = false;

            {
                std::lock_guard<std::mutex> cb_lock(parent_pdp_->callback_mtx_);
                ParticipantDiscoveryInfo info(*old_data);
                info.status = ParticipantDiscoveryInfo::CHANGED_QOS_PARTICIPANT;

                listener->onParticipantDiscovery(
                    parent_pdp_->getRTPSParticipant()->getUserRTPSParticipant(),
                    std::move(info),
                    should_be_ignored);
            }
            if (should_be_ignored)
            {
                parent_pdp_->getRTPSParticipant()->ignore_participant(participant_guid.guidPrefix);
            }
        }
    }

#ifdef FASTDDS_STATISTICS
    //! Addition or update of a participant proxy should trigger
    //! a connections update on the local participant connection list
    if (nullptr != parent_pdp_->getRTPSParticipant()->get_connections_observer())
    {
        parent_pdp_->getRTPSParticipant()->get_connections_observer()->on_local_entity_connections_change(
            parent_pdp_->getRTPSParticipant()->getGuid());
    }
#endif //FASTDDS_STATISTICS

    // Take again the reader lock
    reader->getMutex().lock();
}

    endpoints.stateless_reader.listener_.reset(new PDPSecurityInitiatorListener(this,
            endpoints.stateless_reader.listener_.get(), c_EntityId_SPDPReader, true, false))
    endpoints.reader.listener_.reset(new PDPServerListener(this));
    if (mp_RTPSParticipant->createReader(&reader, ratt, endpoints.reader.history_.get(),
            endpoints.reader.listener_.get(), reader_entity, true, false))
                builtin_endpoints_->main_listener()->onNewCacheChangeAdded(endpoints->reader.reader_, change_aux);
#include <rtps/builtin/discovery/participant/DS/PDPSecurityInitiatorListener.hpp>
#include <rtps/builtin/discovery/participant/simple/SimplePDPEndpointsSecure.hpp>
static HistoryAttributes pdp_reader_history_attributes(
        const BuiltinAttributes& builtin_att,
        const RTPSParticipantAllocationAttributes& allocation)
{
    HistoryAttributes hatt;
    hatt.payloadMaxSize = builtin_att.readerPayloadSize;
    hatt.memoryPolicy = builtin_att.readerHistoryMemoryPolicy;
    hatt.initialReservedCaches = 25;
    if (allocation.participants.initial > 0)
    {
        hatt.initialReservedCaches = (int32_t)allocation.participants.initial;
    }
    if (allocation.participants.maximum < std::numeric_limits<size_t>::max())
    {
        hatt.maximumReservedCaches = (int32_t)allocation.participants.maximum;
    }

    return hatt;
}

static HistoryAttributes pdp_writer_history_attributes(
        const BuiltinAttributes& builtin_att)
{
    HistoryAttributes hatt;
    hatt.payloadMaxSize = builtin_att.writerPayloadSize;
    hatt.memoryPolicy = builtin_att.writerHistoryMemoryPolicy;
    hatt.initialReservedCaches = 1;
    hatt.maximumReservedCaches = 1;

    return hatt;
}
        new_change |= m_hasChangedLocalPDP.exchange(false);
#if HAVE_SECURITY
        if (mp_RTPSParticipant->is_secure())
        {
            auto secure = dynamic_cast<fastdds::rtps::SimplePDPEndpointsSecure*>(builtin_endpoints_.get());
            assert(nullptr != secure);

            RTPSWriter& writer = *(secure->secure_writer.writer_);
            WriterHistory& history = *(secure->secure_writer.history_);
            PDP::announceParticipantState(writer, history, new_change, dispose, wp);
        }
#endif // HAVE_SECURITY

        auto endpoints = dynamic_cast<fastdds::rtps::SimplePDPEndpoints*>(builtin_endpoints_.get());
        RTPSWriter& writer = *(endpoints->writer.writer_);
        WriterHistory& history = *(endpoints->writer.history_);
            endpoints->writer.writer_->unsent_changes_reset();
    fastdds::rtps::SimplePDPEndpoints* endpoints = nullptr;
#if HAVE_SECURITY
    fastdds::rtps::SimplePDPEndpointsSecure* secure_endpoints = nullptr;
    bool is_secure = mp_RTPSParticipant->is_secure();
    if (is_secure)
    {
        secure_endpoints = new fastdds::rtps::SimplePDPEndpointsSecure();
        secure_endpoints->secure_reader.listener_.reset(new PDPListener(this));

        endpoints = secure_endpoints;
        endpoints->reader.listener_.reset(new PDPSecurityInitiatorListener(this));
    }
    else
#endif  // HAVE_SECURITY
    {
        endpoints = new fastdds::rtps::SimplePDPEndpoints();
        endpoints->reader.listener_.reset(new PDPListener(this));
    }
    builtin_endpoints_.reset(endpoints);

    bool ret_val = create_dcps_participant_endpoints();
#if HAVE_SECURITY
    if (ret_val && is_secure)
    {
        create_dcps_participant_secure_endpoints();
    }
#endif  // HAVE_SECURITY
    EPROSIMA_LOG_INFO(RTPS_PDP, "SPDP Endpoints creation finished");
    return ret_val;
}

bool PDPSimple::create_dcps_participant_endpoints()
{
    auto endpoints = dynamic_cast<fastdds::rtps::SimplePDPEndpoints*>(builtin_endpoints_.get());
    assert(nullptr != endpoints);
    constexpr const char* topic_name = "DCPSParticipant";
    const EntityId_t reader_entity_id = c_EntityId_SPDPReader;
    const EntityId_t writer_entity_id = c_EntityId_SPDPWriter;
    // BUILTIN DCPSParticipant READER
    auto& reader = endpoints->reader;
    hatt = pdp_reader_history_attributes(builtin_att, allocation);
    reader.payload_pool_ = TopicPayloadPoolRegistry::get(topic_name, reader_pool_cfg);
    reader.payload_pool_->reserve_history(reader_pool_cfg, true);
    reader.history_.reset(new ReaderHistory(hatt));

    ReaderAttributes ratt = create_builtin_reader_attributes();

    RTPSReader* rtps_reader = nullptr;
    if (mp_RTPSParticipant->createReader(&rtps_reader, ratt, reader.payload_pool_, reader.history_.get(),
            reader.listener_.get(), reader_entity_id, true, false))
    {
        reader.reader_ = dynamic_cast<StatelessReader*>(rtps_reader);
        assert(nullptr != reader.reader_);

        mp_RTPSParticipant->set_endpoint_rtps_protection_supports(rtps_reader, false);
        EPROSIMA_LOG_ERROR(RTPS_PDP, "'" << topic_name << "' builtin reader creation failed");
        reader.release();
    // BUILTIN DCPSParticipant WRITER
    auto& writer = endpoints->writer;
    hatt = pdp_writer_history_attributes(builtin_att);
    writer.payload_pool_ = TopicPayloadPoolRegistry::get(topic_name, writer_pool_cfg);
    writer.payload_pool_->reserve_history(writer_pool_cfg, false);
    writer.history_.reset(new WriterHistory(hatt));

    WriterAttributes watt = create_builtin_writer_attributes();
    RTPSWriter* rtps_writer = nullptr;
    if (mp_RTPSParticipant->createWriter(&rtps_writer, watt, writer.payload_pool_, writer.history_.get(),
            nullptr, writer_entity_id, true))
        writer.writer_ = dynamic_cast<StatelessWriter*>(rtps_writer);
        assert(nullptr != writer.writer_);

        mp_RTPSParticipant->set_endpoint_rtps_protection_supports(rtps_writer, false);

        const NetworkFactory& network = mp_RTPSParticipant->network_factory();
        LocatorList_t fixed_locators;
        for (const Locator_t& loc : mp_builtin->m_initialPeersList)
            if (network.is_locator_remote_or_allowed(loc))
                // Add initial peers locator without transformation as we don't know whether the
                // remote transport will allow localhost
                fixed_locators.push_back(loc);

                /**
                 * TCP special case:
                 *
                 * In TCP, it is not possible to open a socket with 'any' (0.0.0.0) address as it's done
                 * in UDP, so when the TCP transports receive a locator with 'any', they open an input
                 * channel for the specified port in each of the machine interfaces (with the exception
                 * of localhost). In fact, a participant with a TCP transport will only listen on localhost
                 * if localhost is the address of any of the initial peers.
                 *
                 * However, when the TCP enabled participant does not have a whitelist (or localhost is in
                 * it), it allows for transformation of its locators to localhost for performance optimizations.
                 * In this case, the remote TCP participant it will send data using a socket in localhost,
                 * and for that the participant with the initial peers list needs to be listening there
                 * to receive it.
                 *
                 * That means:
                 *   1. Checking that the initial peer is not already localhost
                 *   2. Checking that the initial peer locator is of TCP kind
                 *   3. Checking that the network configuration allows for localhost locators
                 */
                Locator_t local_locator;
                network.transform_remote_locator(loc, local_locator,
                        DISC_NETWORK_CONFIGURATION_LISTENING_LOCALHOST_ALL);
                if (loc != local_locator
                        && (loc.kind == LOCATOR_KIND_TCPv4 || loc.kind == LOCATOR_KIND_TCPv6)
                        && network.is_locator_allowed(local_locator))
                    fixed_locators.push_back(local_locator);
            else
            {
                EPROSIMA_LOG_WARNING(RTPS_PDP, "Ignoring initial peers locator " << loc << " : not allowed.");
            }
        writer.writer_->set_fixed_locators(fixed_locators);
        EPROSIMA_LOG_ERROR(RTPS_PDP, "'" << topic_name << "' builtin writer creation failed");
        writer.release();
#if HAVE_SECURITY
bool PDPSimple::create_dcps_participant_secure_endpoints()
    const RTPSParticipantAttributes& pattr = mp_RTPSParticipant->getRTPSParticipantAttributes();
    const RTPSParticipantAllocationAttributes& allocation = pattr.allocation;
    const BuiltinAttributes& builtin_att = mp_builtin->m_att;
    auto endpoints = dynamic_cast<fastdds::rtps::SimplePDPEndpointsSecure*>(builtin_endpoints_.get());
    assert(nullptr != endpoints);
    constexpr const char* topic_name = "DCPSParticipantsSecure";
    const EntityId_t reader_entity_id = c_EntityId_spdp_reliable_participant_secure_reader;
    const EntityId_t writer_entity_id = c_EntityId_spdp_reliable_participant_secure_writer;
    // BUILTIN DCPSParticipantsSecure READER
    auto& reader = endpoints->secure_reader;
    HistoryAttributes hatt;
    hatt = pdp_reader_history_attributes(builtin_att, allocation);
    PoolConfig reader_pool_cfg = PoolConfig::from_history_attributes(hatt);
    reader.payload_pool_ = TopicPayloadPoolRegistry::get(topic_name, reader_pool_cfg);
    reader.payload_pool_->reserve_history(reader_pool_cfg, true);
    reader.history_.reset(new ReaderHistory(hatt));

    ReaderAttributes ratt = create_builtin_reader_attributes();
    WriterAttributes watt = create_builtin_writer_attributes();
    add_builtin_security_attributes(ratt, watt);

    RTPSReader* rtps_reader = nullptr;
    if (mp_RTPSParticipant->createReader(&rtps_reader, ratt, reader.payload_pool_, reader.history_.get(),
            reader.listener_.get(), reader_entity_id, true, false))
    {
        reader.reader_ = dynamic_cast<StatefulReader*>(rtps_reader);
        assert(nullptr != reader.reader_);
    else
        EPROSIMA_LOG_ERROR(RTPS_PDP, "'" << topic_name << "' builtin reader creation failed");
        reader.release();
        return false;
    }
    // SPDP BUILTIN RTPSParticipant WRITER
    auto& writer = endpoints->secure_writer;
    hatt = pdp_writer_history_attributes(builtin_att);
    PoolConfig writer_pool_cfg = PoolConfig::from_history_attributes(hatt);
    writer.payload_pool_ = TopicPayloadPoolRegistry::get(topic_name, writer_pool_cfg);
    writer.payload_pool_->reserve_history(writer_pool_cfg, false);
    writer.history_.reset(new WriterHistory(hatt));
    RTPSWriter* rtps_writer = nullptr;
    if (mp_RTPSParticipant->createWriter(&rtps_writer, watt, writer.payload_pool_, writer.history_.get(),
            nullptr, writer_entity_id, true))
    {
        writer.writer_ = dynamic_cast<StatefulWriter*>(rtps_writer);
        assert(nullptr != writer.writer_);
    }
    else
    {
        EPROSIMA_LOG_ERROR(RTPS_PDP, "'" << topic_name << "' builtin writer creation failed");
        writer.release();
        return false;
    }
    return true;
}

#endif  // HAVE_SECURITY

void PDPSimple::assignRemoteEndpoints(
        ParticipantProxyData* pdata)
{
    bool ignored = false;
    notify_and_maybe_ignore_new_participant(pdata, ignored);
    if (!ignored)
    {
#if HAVE_SECURITY
        auto endpoints = dynamic_cast<fastdds::rtps::SimplePDPEndpointsSecure*>(builtin_endpoints_.get());
        if (nullptr != endpoints)
            // This participant is secure.
            // PDP should have been matched inside notifyAboveRemoteEndpoints after completing the authentication process.
            // We now match the other builtin endpoints.
            GUID_t remote_guid = pdata->m_guid;
            remote_guid.entityId = c_EntityId_spdp_reliable_participant_secure_writer;
            bool notify_secure = endpoints->secure_reader.reader_->matched_writer_is_matched(remote_guid);
            assign_low_level_remote_endpoints(*pdata, notify_secure);
#endif // if HAVE_SECURITY
            // This participant is not secure.
            // Match PDP and other builtin endpoints.
            match_pdp_remote_endpoints(*pdata, false);
            assign_low_level_remote_endpoints(*pdata, false);
    GUID_t guid = pdata->m_guid;
        auto endpoints = dynamic_cast<fastdds::rtps::SimplePDPEndpoints*>(builtin_endpoints_.get());
        assert(nullptr != endpoints);

        guid.entityId = c_EntityId_SPDPWriter;
        endpoints->reader.reader_->matched_writer_remove(guid);

        guid.entityId = c_EntityId_SPDPReader;
        endpoints->writer.writer_->matched_reader_remove(guid);

#if HAVE_SECURITY
    auto endpoints = dynamic_cast<fastdds::rtps::SimplePDPEndpointsSecure*>(builtin_endpoints_.get());
    if (nullptr != endpoints)
        guid.entityId = c_EntityId_spdp_reliable_participant_secure_writer;
        endpoints->secure_reader.reader_->matched_writer_remove(guid);

        guid.entityId = c_EntityId_spdp_reliable_participant_secure_reader;
        endpoints->secure_writer.writer_->matched_reader_remove(guid);
#endif // HAVE_SECURITY
    if (notify_secure_endpoints)
    {
        match_pdp_remote_endpoints(pdata, true);
    }
    else
    {
        // Add remote participant data
        GUID_t writer_guid{ pdata.m_guid.guidPrefix, c_EntityId_SPDPWriter };
        ParticipantProxyData* part_data = createParticipantProxyData(pdata, writer_guid);
        if (part_data != nullptr)
        {
            bool ignored = false;
            notify_and_maybe_ignore_new_participant(part_data, ignored);
            if (!ignored)
            {
                match_pdp_remote_endpoints(*part_data, false);
                assign_low_level_remote_endpoints(*part_data, false);
            }
        }
    }

}

void PDPSimple::match_pdp_remote_endpoints(
        const ParticipantProxyData& pdata,
        bool notify_secure_endpoints)
{
#if !HAVE_SECURITY
    static_cast<void>(notify_secure_endpoints);
#endif // !HAVE_SECURITY

    auto endpoints = static_cast<fastdds::rtps::SimplePDPEndpoints*>(builtin_endpoints_.get());

    const NetworkFactory& network = mp_RTPSParticipant->network_factory();
    bool use_multicast_locators = !mp_RTPSParticipant->getAttributes().builtin.avoid_builtin_multicast ||
            pdata.metatraffic_locators.unicast.empty();
    const uint32_t endp = pdata.m_availableBuiltinEndpoints;

    // Default to values for non-secure endpoints
    auto reliability_kind = BEST_EFFORT_RELIABILITY_QOS;
    uint32_t pdp_reader_mask = DISC_BUILTIN_ENDPOINT_PARTICIPANT_DETECTOR;
    uint32_t pdp_writer_mask = DISC_BUILTIN_ENDPOINT_PARTICIPANT_ANNOUNCER;
    EntityId_t reader_entity_id = c_EntityId_SPDPReader;
    EntityId_t writer_entity_id = c_EntityId_SPDPWriter;
    RTPSReader* reader = endpoints->reader.reader_;
    RTPSWriter* writer = endpoints->writer.writer_;

#if HAVE_SECURITY
    // If the other participant has been authenticated, use values for secure endpoints
    if (notify_secure_endpoints)
    {
        auto secure_endpoints = static_cast<fastdds::rtps::SimplePDPEndpointsSecure*>(builtin_endpoints_.get());
        reliability_kind = RELIABLE_RELIABILITY_QOS;
        pdp_reader_mask = DISC_BUILTIN_ENDPOINT_PARTICIPANT_SECURE_DETECTOR;
        pdp_writer_mask = DISC_BUILTIN_ENDPOINT_PARTICIPANT_SECURE_ANNOUNCER;
        reader_entity_id = c_EntityId_spdp_reliable_participant_secure_reader;
        writer_entity_id = c_EntityId_spdp_reliable_participant_secure_writer;
        reader = secure_endpoints->secure_reader.reader_;
        writer = secure_endpoints->secure_writer.writer_;
    }
#endif // HAVE_SECURITY

    if (0 != (endp & pdp_writer_mask))
    {
        auto temp_writer_data = get_temporary_writer_proxies_pool().get();

        temp_writer_data->clear();
        temp_writer_data->guid().guidPrefix = pdata.m_guid.guidPrefix;
        temp_writer_data->guid().entityId = writer_entity_id;
        temp_writer_data->persistence_guid(pdata.get_persistence_guid());
        temp_writer_data->set_persistence_entity_id(writer_entity_id);
        temp_writer_data->set_remote_locators(pdata.metatraffic_locators, network, use_multicast_locators);
        temp_writer_data->m_qos.m_reliability.kind = reliability_kind;
        temp_writer_data->m_qos.m_durability.kind = TRANSIENT_LOCAL_DURABILITY_QOS;
#if HAVE_SECURITY
        if (notify_secure_endpoints)
        {
            if (!mp_RTPSParticipant->security_manager().discovered_builtin_writer(
                        reader->getGuid(), pdata.m_guid, *temp_writer_data,
                        reader->getAttributes().security_attributes()))
            {
                EPROSIMA_LOG_ERROR(RTPS_EDP, "Security manager returns an error for writer " <<
                        temp_writer_data->guid());
            }
        }
        else
#endif // HAVE_SECURITY
        {
            reader->matched_writer_add(*temp_writer_data);
        }
    }

    if (0 != (endp & pdp_reader_mask))
    {
        auto temp_reader_data = get_temporary_reader_proxies_pool().get();

        temp_reader_data->clear();
        temp_reader_data->m_expectsInlineQos = false;
        temp_reader_data->guid().guidPrefix = pdata.m_guid.guidPrefix;
        temp_reader_data->guid().entityId = reader_entity_id;
        temp_reader_data->set_remote_locators(pdata.metatraffic_locators, network, use_multicast_locators);
        temp_reader_data->m_qos.m_reliability.kind = reliability_kind;
        temp_reader_data->m_qos.m_durability.kind = TRANSIENT_LOCAL_DURABILITY_QOS;
#if HAVE_SECURITY
        if (notify_secure_endpoints)
        {
            if (!mp_RTPSParticipant->security_manager().discovered_builtin_reader(
                        writer->getGuid(), pdata.m_guid, *temp_reader_data,
                        writer->getAttributes().security_attributes()))
            {
                EPROSIMA_LOG_ERROR(RTPS_EDP, "Security manager returns an error for reader " <<
                        temp_reader_data->guid());
            }
        }
        else
#endif // HAVE_SECURITY
        {
            writer->matched_reader_add(*temp_reader_data);
        }

        if (BEST_EFFORT_RELIABILITY_QOS == reliability_kind)
        {
            endpoints->writer.writer_->unsent_changes_reset();
        }
    }
}

void PDPSimple::assign_low_level_remote_endpoints(
        const ParticipantProxyData& pdata,
        bool notify_secure_endpoints)
{
        mp_EDP->assignRemoteEndpoints(pdata, notify_secure_endpoints);
        mp_builtin->mp_WLP->assignRemoteEndpoints(pdata, notify_secure_endpoints);
#if HAVE_SECURITY
bool PDPSimple::pairing_remote_writer_with_local_reader_after_security(
        const GUID_t& local_reader,
        const WriterProxyData& remote_writer_data)
{
    auto endpoints = dynamic_cast<fastdds::rtps::SimplePDPEndpointsSecure*>(builtin_endpoints_.get());
    if ((nullptr != endpoints) && (local_reader == endpoints->secure_reader.reader_->getGuid()))
    {
        endpoints->secure_reader.reader_->matched_writer_add(remote_writer_data);
        return true;
    }

    return PDP::pairing_remote_writer_with_local_reader_after_security(local_reader, remote_writer_data);
}

bool PDPSimple::pairing_remote_reader_with_local_writer_after_security(
        const GUID_t& local_writer,
        const ReaderProxyData& remote_reader_data)
{
    auto endpoints = dynamic_cast<fastdds::rtps::SimplePDPEndpointsSecure*>(builtin_endpoints_.get());
    if ((nullptr != endpoints) && (local_writer == endpoints->secure_writer.writer_->getGuid()))
    {
        endpoints->secure_writer.writer_->matched_reader_add(remote_reader_data);
        return true;
    }

    return PDP::pairing_remote_reader_with_local_writer_after_security(local_writer, remote_reader_data);
}

#endif // HAVE_SECURITY

    const std::unique_ptr<fastrtps::rtps::ReaderListener>& main_listener() const override
    {
        return reader.listener_;
    }

// Copyright 2023 Proyectos y Sistemas de Mantenimiento SL (eProsima).
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unful required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

/**
 * @file SimplePDPEndpointsSecure.hpp
 */

#ifndef FASTDDS_RTPS_BUILTIN_DISCOVERY_PARTICIPANT_SIMPLE__SIMPLEPDPENDPOINTSSECURE_HPP_
#define FASTDDS_RTPS_BUILTIN_DISCOVERY_PARTICIPANT_SIMPLE__SIMPLEPDPENDPOINTSSECURE_HPP_

#include <memory>

#include <fastdds/rtps/builtin/data/BuiltinEndpoints.hpp>
#include <fastdds/rtps/reader/StatefulReader.h>
#include <fastdds/rtps/writer/StatefulWriter.h>

#include <rtps/builtin/BuiltinReader.hpp>
#include <rtps/builtin/BuiltinWriter.hpp>
#include <rtps/builtin/discovery/participant/simple/SimplePDPEndpoints.hpp>
#include <rtps/history/ITopicPayloadPool.h>

namespace eprosima {
namespace fastdds {
namespace rtps {

/**
 * Container for the builtin endpoints of SPDPSimple
 */
struct SimplePDPEndpointsSecure : public SimplePDPEndpoints
{
    ~SimplePDPEndpointsSecure() override = default;

    fastrtps::rtps::BuiltinEndpointSet_t builtin_endpoints() const override
    {
        return SimplePDPEndpoints::builtin_endpoints() |
               DISC_BUILTIN_ENDPOINT_PARTICIPANT_SECURE_ANNOUNCER | DISC_BUILTIN_ENDPOINT_PARTICIPANT_SECURE_DETECTOR;
    }

    const std::unique_ptr<fastrtps::rtps::ReaderListener>& main_listener() const override
    {
        return secure_reader.listener_;
    }

    bool enable_pdp_readers(
            fastrtps::rtps::RTPSParticipantImpl* participant) override
    {
        return SimplePDPEndpoints::enable_pdp_readers(participant) &&
               participant->enableReader(secure_reader.reader_);
    }

    void disable_pdp_readers(
            fastrtps::rtps::RTPSParticipantImpl* participant) override
    {
        participant->disableReader(secure_reader.reader_);
        SimplePDPEndpoints::disable_pdp_readers(participant);
    }

    void delete_pdp_endpoints(
            fastrtps::rtps::RTPSParticipantImpl* participant) override
    {
        participant->deleteUserEndpoint(secure_reader.reader_->getGuid());
        participant->deleteUserEndpoint(secure_writer.writer_->getGuid());
        SimplePDPEndpoints::delete_pdp_endpoints(participant);
    }

    void remove_from_pdp_reader_history(
            const fastrtps::rtps::InstanceHandle_t& remote_participant) override
    {
        secure_reader.remove_from_history(remote_participant);
        SimplePDPEndpoints::remove_from_pdp_reader_history(remote_participant);
    }

    void remove_from_pdp_reader_history(
            fastrtps::rtps::CacheChange_t* change) override
    {
        assert(nullptr != change);
        if (change->writerGUID.entityId == fastrtps::rtps::c_EntityId_SPDPWriter)
        {
            secure_reader.history_->remove_change(change);
        }
        else
        {
            SimplePDPEndpoints::remove_from_pdp_reader_history(change);
        }
    }

    //! Builtin Simple PDP secure reader
    BuiltinReader<fastrtps::rtps::StatefulReader> secure_reader;

    //! Builtin Simple PDP secure writer
    BuiltinWriter<fastrtps::rtps::StatefulWriter> secure_writer;
};

} // namespace rtps
} // namespace fastdds
} // namespace eprosima

#endif  // FASTDDS_RTPS_BUILTIN_DISCOVERY_PARTICIPANT_SIMPLE__SIMPLEPDPENDPOINTSSECURE_HPP_
        bool update_notified = true;

        decltype(matched_writers_)::iterator writer = matched_writers_.end();
        if ((EXCLUSIVE_OWNERSHIP_QOS == m_att.ownershipKind) ||
                (m_trustedWriterEntityId == change->writerGUID.entityId))
            writer = std::find_if(matched_writers_.begin(), matched_writers_.end(),
            bool is_matched = matched_writers_.end() != writer;
            update_notified = is_matched;
        }

        // Update Ownership strength.
        if (EXCLUSIVE_OWNERSHIP_QOS == m_att.ownershipKind)
        {
            SequenceNumber_t previous_seq{ 0, 0 };
            if (update_notified)
            {
                previous_seq = update_last_notified(change->writerGUID, change->sequenceNumber);
            }

        if (remote_participant_info->auth_status_ == AUTHENTICATION_FAILED)
        {
            remote_participant_info->auth_status_ = AUTHENTICATION_REQUEST_NOT_SEND;
        }
        remote_participant_info->auth_status_ = AUTHENTICATION_FAILED;
                if (remote_participant_info->auth_status_ != AUTHENTICATION_FAILED)
                {
                    SecurityException exception;
                    remote_participant_info->event_->cancel_timer();
                    remote_participant_info->auth_status_ = AUTHENTICATION_FAILED;
                    on_validation_failed(dp_it->second->participant_data(), exception);
                }
#include <atomic>
#include <algorithm>
#include <fstream>
#include <map>

#include <gtest/gtest.h>

#include <fastdds/dds/log/Log.hpp>
#include <fastdds/rtps/common/EntityId_t.hpp>
#include <fastrtps/xmlparser/XMLProfileManager.h>
#include <rtps/transport/test_UDPv4Transport.h>
struct UDPMessageSender
{
    asio::io_service service;
    asio::ip::udp::socket socket;

    UDPMessageSender()
        : service()
        , socket(service)
    {
        socket.open(asio::ip::udp::v4());
    }

    void send(
            const CDRMessage_t& msg,
            const Locator_t& destination)
    {
        std::string addr = IPLocator::toIPv4string(destination);
        unsigned short port = static_cast<unsigned short>(destination.port);
        auto remote = asio::ip::udp::endpoint(asio::ip::address::from_string(addr), port);
        asio::error_code ec;

        socket.send_to(asio::buffer(msg.buffer, msg.length), remote, 0, ec);
    }

};
TEST(Security, BuiltinAuthenticationPlugin_second_participant_creation_loop)
    // Class to allow waiting for the authentication message to be sent
    class AuthMessageSendStatus
    {
        bool message_sent_ = false;
        std::mutex mutex_;
        std::condition_variable cv_;

    public:

        void reset()
        {
            std::lock_guard < std::mutex> guard(mutex_);
            message_sent_ = false;
        }

        void notify()
        {
            std::lock_guard<std::mutex> guard(mutex_);
            message_sent_ = true;
            cv_.notify_one();
        }

        void wait()
        {
            std::unique_lock<std::mutex> lock(mutex_);
            cv_.wait(lock, [this]() -> bool
                    {
                        return message_sent_;
                    });
        }

    };

    // Prepare transport to check that the authentication message is sent
    auto transport = std::make_shared<test_UDPv4TransportDescriptor>();
    AuthMessageSendStatus auth_message_send_status;
    transport->drop_data_messages_filter_ = [&auth_message_send_status](eprosima::fastrtps::rtps::CDRMessage_t& msg)
            -> bool
            {
                auto old_pos = msg.pos;

                // Jump to writer entity id
                msg.pos += 2 + 2 + 4;

                // Read writer entity id
                eprosima::fastrtps::rtps::GUID_t writer_guid;
                eprosima::fastrtps::rtps::CDRMessage::readEntityId(&msg, &writer_guid.entityId);
                msg.pos = old_pos;

                if (writer_guid.entityId == eprosima::fastrtps::rtps::participant_stateless_message_writer_entity_id)
                {
                    auth_message_send_status.notify();
                }

                return false;
            };

    main_participant.disable_builtin_transport().add_user_transport_to_pparams(transport);
        auth_message_send_status.reset();
        // Wait for the main participant to send an authentication message to the other participant
        auth_message_send_status.wait();
    //! Wait for the authorization to fail (~15secs)
    writer.waitUnauthorized();

    //! Wait for the discovery
    writer.wait_discovery();
    ASSERT_TRUE(writer.is_matched());
    //! Wait for the authorization to fail (~15secs)
    writer.waitUnauthorized();

    //! Wait some time afterwards (this will time out)
    writer.wait_discovery(std::chrono::seconds(1));
    ASSERT_FALSE(writer.is_matched());
    UDPMessageSender fake_msg_sender;
        fake_msg_sender.send(msg, reader_locator);
TEST_P(Security, MaliciousParticipantRemovalIgnore)
{
    PubSubWriter<HelloWorldPubSubType> writer("HelloWorldTopic_MaliciousParticipantRemovalIgnore");
    PubSubReader<HelloWorldPubSubType> reader("HelloWorldTopic_MaliciousParticipantRemovalIgnore");

    struct MaliciousParticipantRemoval
    {
        std::array<char, 4> rtps_id{ {'R', 'T', 'P', 'S'} };
        std::array<uint8_t, 2> protocol_version{ {2, 3} };
        std::array<uint8_t, 2> vendor_id{ {0x01, 0x0F} };
        GuidPrefix_t sender_prefix{};

        struct DataSubMsg
        {
            struct Header
            {
                uint8_t submessage_id = 0x15;
#if FASTDDS_IS_BIG_ENDIAN_TARGET
                uint8_t flags = 0x02;
#else
                uint8_t flags = 0x03;
#endif  // FASTDDS_IS_BIG_ENDIAN_TARGET
                uint16_t submessage_length = 2 + 2 + 4 + 4 + 8;
                uint16_t extra_flags = 0;
                uint16_t octets_to_inline_qos = 4 + 4 + 8;
                EntityId_t reader_id{};
                EntityId_t writer_id{};
                SequenceNumber_t sn{};
            };

            struct InlineQos
            {
                struct KeyHash
                {
                    uint16_t pid = 0x0070;  // PID_KEY_HASH
                    uint16_t plen = 16;
                    GUID_t guid{};
                };

                struct StatusInfo
                {
                    uint16_t pid = 0x0071;  // PID_STATUS_INFO
                    uint16_t plen = 4;
                    uint8_t flags[4] = { 0x00, 0x00, 0x00, 0x03 };
                };

                struct Sentinel
                {
                    uint16_t pid = 0x0001;  // PID_SENTINEL
                    uint16_t plen = 0;
                };

                KeyHash hash;
                StatusInfo status;
                Sentinel sentinel;
            };

            Header header;
            InlineQos inline_qos;
        }
        data;
    };

    // Set common QoS
    reader.history_depth(10).reliability(eprosima::fastrtps::RELIABLE_RELIABILITY_QOS);
    writer.history_depth(10).reliability(eprosima::fastrtps::RELIABLE_RELIABILITY_QOS);

    // Configure security
    const std::string governance_file("governance_helloworld_all_enable.smime");
    const std::string permissions_file("permissions_helloworld.smime");
    CommonPermissionsConfigure(reader, writer, governance_file, permissions_file);

    // Initialize and wait for authorization and discovery
    reader.init();
    ASSERT_TRUE(reader.isInitialized());
    writer.init();
    ASSERT_TRUE(writer.isInitialized());
    reader.waitAuthorized();
    writer.waitAuthorized();
    reader.wait_discovery();
    writer.wait_discovery();

    // Send fake DATA(p[UD])
    UDPMessageSender fake_msg_sender;
    {
        auto writer_guid = writer.datawriter_guid();
        auto participant_guid = writer.participant_guid();
        auto domain_id = static_cast<uint32_t>(GET_PID() % 230);

        MaliciousParticipantRemoval packet{};
        packet.sender_prefix = writer_guid.guidPrefix;
        packet.data.header.submessage_length += sizeof(packet.data.inline_qos);
        packet.data.header.writer_id = c_EntityId_SPDPWriter;
        packet.data.header.reader_id = c_EntityId_SPDPReader;
        packet.data.header.sn.low = 100;
        packet.data.inline_qos.hash.guid = participant_guid;

        Locator_t mcast_locator;
        ASSERT_TRUE(IPLocator::setIPv4(mcast_locator, "239.255.0.1"));
        mcast_locator.port = 7400 + 250 * domain_id;

        CDRMessage_t msg(0);
        uint32_t msg_len = static_cast<uint32_t>(sizeof(packet));
        msg.init(reinterpret_cast<octet*>(&packet), msg_len);
        msg.length = msg_len;
        msg.pos = msg_len;
        fake_msg_sender.send(msg, mcast_locator);
    }

    EXPECT_FALSE(reader.wait_participant_undiscovery(std::chrono::seconds(1)));

    auto data = default_helloworld_data_generator();
    reader.startReception(data);
    writer.send(data);
    ASSERT_TRUE(data.empty());
    reader.block_for_all();
}


    MOCK_METHOD(bool, ignore_participant, (const GuidPrefix_t&));

#include <memory>
#include <fastdds/rtps/reader/ReaderListener.h>
    const std::unique_ptr<fastrtps::rtps::ReaderListener>& main_listener() const override
    {
        return no_listener_;
    }

    std::unique_ptr<fastrtps::rtps::ReaderListener> no_listener_;