[21696] Fix Secure Discovery Server client disposals guid and handshake_handle assertion

src/cpp/rtps/builtin/discovery/participant/PDPClient.cpp

@@ -769,9 +769,17 @@ void PDPClient::announceParticipantState(
                         // if we are matched to a server report demise
                         if (svr.is_connected)
                         {
+                            GuidPrefix_t srv_guid_prefix = svr.guidPrefix;
+
+                            if (getRTPSParticipant()->is_secure())
+                            {
+                                // Need the mangled guid prefix in this case
+                                srv_guid_prefix = get_participant_proxy_data(svr.guidPrefix)->m_guid.guidPrefix;
+                            }
+
                             //locators.push_back(svr.metatrafficMulticastLocatorList);
                             locators.push_back(svr.metatrafficUnicastLocatorList);
-                            remote_readers.emplace_back(svr.guidPrefix,
+                            remote_readers.emplace_back(srv_guid_prefix,
                                     endpoints->reader.reader_->getGuid().entityId);
                         }
                     }

src/cpp/rtps/security/SecurityManager.cpp

@@ -935,6 +935,10 @@ bool SecurityManager::on_process_handshake(
                 }
                 else
                 {
+                    // Return the handshake handle
+                    authentication_plugin_->return_handshake_handle(remote_participant_info->handshake_handle_,
+                            exception);
+                    remote_participant_info->handshake_handle_ = nullptr;
                     EPROSIMA_LOG_ERROR(SECURITY, "WriterHistory cannot add the CacheChange_t");
                 }
             }
@@ -946,6 +950,9 @@ bool SecurityManager::on_process_handshake(
         }
         else
         {
+            // Return the handshake handle
+            authentication_plugin_->return_handshake_handle(remote_participant_info->handshake_handle_, exception);
+            remote_participant_info->handshake_handle_ = nullptr;
             EPROSIMA_LOG_ERROR(SECURITY, "WriterHistory cannot retrieve a CacheChange_t");
         }
     }

test/dds/communication/security/CMakeLists.txt

@@ -18,7 +18,7 @@ add_definitions(
     -DBOOST_ASIO_STANDALONE
     -DASIO_STANDALONE
     )
-    
+
 include_directories(${Asio_INCLUDE_DIR})
 
 ###############################################################################
@@ -393,6 +393,48 @@ if(SECURITY)
                 "PATH=$<TARGET_FILE_DIR:${PROJECT_NAME}>\\;$<TARGET_FILE_DIR:fastcdr>\\;${WIN_PATH}")
         endif()
 
+        add_test(NAME SecureDiscoverServerMultipleClientsHandShakeAssertion
+            COMMAND ${PYTHON_EXECUTABLE} ${CMAKE_CURRENT_BINARY_DIR}/multiple_secure_ds_pubsub_secure_crypto_communication.py
+            --pub $<TARGET_FILE:CommunicationPublisher>
+            --xml-pub secure_ds_simple_secure_msg_crypto_pub_profile.xml
+            --sub $<TARGET_FILE:CommunicationSubscriber>
+            --xml-sub secure_ds_simple_secure_msg_crypto_sub_profile.xml
+            --samples 100 #not important to receive all samples
+            --servers $<TARGET_FILE:fast-discovery-server>
+            --xml-servers secure_simple_ds_server_profile.xml
+            --n-clients 30
+            --relaunch-clients
+            --validation-method server)
+
+        # Set test with label NoMemoryCheck
+        set_property(TEST SecureDiscoverServerMultipleClientsHandShakeAssertion PROPERTY LABELS "NoMemoryCheck")
+
+        if(WIN32)
+            string(REPLACE ";" "\\;" WIN_PATH "$ENV{PATH}")
+            set_property(TEST SecureDiscoverServerMultipleClientsHandShakeAssertion APPEND PROPERTY ENVIRONMENT
+                "PATH=$<TARGET_FILE_DIR:${PROJECT_NAME}>\\;$<TARGET_FILE_DIR:fastcdr>\\;${WIN_PATH}")
+        endif()
+
+        add_test(NAME SecureDiscoverServerMultipleClientsDisposalsReceived
+            COMMAND ${PYTHON_EXECUTABLE} ${CMAKE_CURRENT_BINARY_DIR}/multiple_secure_ds_pubsub_secure_crypto_communication.py
+            --pub $<TARGET_FILE:CommunicationPublisher>
+            --xml-pub secure_ds_simple_secure_msg_crypto_pub_profile.xml
+            --sub $<TARGET_FILE:CommunicationSubscriber>
+            --xml-sub secure_ds_simple_secure_msg_crypto_sub_profile.xml
+            --samples 5
+            --servers $<TARGET_FILE:fast-discovery-server>
+            --xml-servers secure_simple_ds_server_profile.xml
+            --exit-on-disposal-received)
+
+        # Set test with label NoMemoryCheck
+        set_property(TEST SecureDiscoverServerMultipleClientsDisposalsReceived PROPERTY LABELS "NoMemoryCheck")
+
+        if(WIN32)
+            string(REPLACE ";" "\\;" WIN_PATH "$ENV{PATH}")
+            set_property(TEST SecureDiscoverServerMultipleClientsDisposalsReceived APPEND PROPERTY ENVIRONMENT
+                "PATH=$<TARGET_FILE_DIR:${PROJECT_NAME}>\\;$<TARGET_FILE_DIR:fastcdr>\\;${WIN_PATH}")
+        endif()
+
     endif()
 
 endif()

test/dds/communication/security/PublisherMain.cpp

@@ -40,6 +40,7 @@ int main(
 {
     int arg_count = 1;
     bool exit_on_lost_liveliness = false;
+    bool exit_on_disposal_received = false;
     bool fixed_type = false;
     bool zero_copy = false;
     uint32_t seed = 7800;
@@ -55,6 +56,10 @@ int main(
         {
             exit_on_lost_liveliness = true;
         }
+        else if (strcmp(argv[arg_count], "--exit_on_disposal_received") == 0)
+        {
+            exit_on_disposal_received = true;
+        }
         else if (strcmp(argv[arg_count], "--fixed_type") == 0)
         {
             fixed_type = true;
@@ -137,7 +142,7 @@ int main(
         DomainParticipantFactory::get_instance()->load_XML_profiles_file(xml_file);
     }
 
-    PublisherModule publisher(exit_on_lost_liveliness, fixed_type, zero_copy);
+    PublisherModule publisher(exit_on_lost_liveliness, exit_on_disposal_received, fixed_type, zero_copy);
 
     if (publisher.init(seed, magic))
     {

test/dds/communication/security/PublisherModule.cpp

@@ -230,6 +230,10 @@ void PublisherModule::on_participant_discovery(
     {
         std::cout << "Publisher participant " << // participant->getGuid() <<
             " removed participant " << info.info.m_guid << std::endl;
+        if (exit_on_disposal_received_)
+        {
+            run_ = false;
+        }
     }
     else if (info.status == ParticipantDiscoveryInfo::DROPPED_PARTICIPANT)
     {

test/dds/communication/security/PublisherModule.hpp

@@ -41,9 +41,11 @@ class PublisherModule
 
     PublisherModule(
             bool exit_on_lost_liveliness,
+            bool exit_on_disposal_received,
             bool fixed_type = false,
             bool zero_copy = false)
         : exit_on_lost_liveliness_(exit_on_lost_liveliness)
+        , exit_on_disposal_received_(exit_on_disposal_received)
         , fixed_type_(zero_copy || fixed_type) // If zero copy active, fixed type is required
         , zero_copy_(zero_copy)
     {
@@ -91,6 +93,7 @@ class PublisherModule
     std::condition_variable cv_;
     unsigned int matched_ = 0;
     bool exit_on_lost_liveliness_ = false;
+    bool exit_on_disposal_received_ = false;
     bool fixed_type_ = false;
     bool zero_copy_ = false;
     bool run_ = true;

test/dds/communication/security/multiple_secure_ds_pubsub_secure_crypto_communication.py

@@ -18,6 +18,7 @@
 import os
 import subprocess
 import sys
+import time
 
 class ParseOptions():
     """Parse arguments."""
@@ -93,9 +94,106 @@ def __parse_args(self):
             type=str,
             help='Path to the xml configuration file containing discovery server.'
         )
+        parser.add_argument(
+            '-nc',
+            '--n-clients',
+            type=int,
+            help='Number of pubsub clients to launch (1 of each).'
+        )
+        parser.add_argument(
+            '-rc',
+            '--relaunch-clients',
+            action='store_true',
+            help='Whether to kill clients and relaunch them.'
+        )
+        parser.add_argument(
+            '-vm',
+            '--validation-method',
+            type=str,
+            help='Validation method to use [server, subscriber].'
+        )
+        parser.add_argument(
+            '-edr',
+            '--exit-on-disposal-received',
+            action='store_true',
+            help='Let the publisher finish the process if receives a disposal.'
+        )
 
         return parser.parse_args()
 
+def launch_discovery_server_processes(servers, xml_servers):
+
+    ds_procs = []
+
+    for i in range(0, len(servers)):
+        server_cmd = []
+
+        if not os.path.isfile(servers[i]):
+            print(f'Discovery server executable file does not exists: {servers[i]}')
+            sys.exit(1)
+
+        if not os.access(servers[i], os.X_OK):
+            print(
+                'Discovery server executable does not have execution permissions:'
+                f'{servers[i]}')
+            sys.exit(1)
+
+        server_cmd.append(servers[i])
+        server_cmd.extend(['--xml-file', xml_servers[i]])
+        server_cmd.extend(['--server-id', str(i)])
+
+        ds_proc = subprocess.Popen(server_cmd)
+        print(
+            'Running Discovery Server - commmand:  ',
+            ' '.join(map(str, server_cmd)))
+
+        ds_procs.append(ds_proc)
+
+    return ds_procs
+
+def launch_client_processes(n_clients, pub_command, sub_command):
+
+    pub_procs = []
+    sub_procs = []
+
+    for i in range(0, n_clients):
+        sub_proc = subprocess.Popen(sub_command)
+        print(
+                f'Running Subscriber - commmand:  ',
+                ' '.join(map(str, sub_command)))
+
+        pub_proc = subprocess.Popen(pub_command)
+        print(
+            'Running Publisher - commmand:  ',
+            ' '.join(map(str, pub_command)))
+
+        sub_procs.append(sub_proc)
+        pub_procs.append(pub_proc)
+
+    return sub_procs, pub_procs
+
+def cleanup(pub_procs, sub_procs, ds_procs):
+
+    [sub_proc.kill() for sub_proc in sub_procs]
+    [pub_proc.kill() for pub_proc in pub_procs]
+    [ds_proc.kill() for ds_proc in ds_procs]
+
+def cleanup_clients(pub_procs, sub_procs):
+
+    [sub_proc.kill() for sub_proc in sub_procs]
+    [pub_proc.kill() for pub_proc in pub_procs]
+
+def terminate(ok = True):
+    if ok:
+        try:
+            sys.exit(os.EX_OK)
+        except AttributeError:
+            sys.exit(0)
+    else:
+        try:
+            sys.exit(os.EX_SOFTWARE)
+        except AttributeError:
+            sys.exit(1)
 
 def run(args):
     """
@@ -108,6 +206,8 @@ def run(args):
     """
     pub_command = []
     sub_command = []
+    n_clients = 1
+    relaunch_clients = False
 
     script_dir = os.path.dirname(os.path.realpath(__file__))
 
@@ -152,75 +252,64 @@ def run(args):
     if args.wait:
         pub_command.extend(['--wait', str(args.wait)])
 
+    if args.exit_on_disposal_received:
+        pub_command.extend(['--exit_on_disposal_received'])
+
     if args.samples:
         pub_command.extend(['--samples', str(args.samples)])
         sub_command.extend(['--samples', str(args.samples)])
 
+    if args.n_clients:
+        n_clients = int(args.n_clients)
+
+    if args.relaunch_clients:
+        relaunch_clients = True
+
     if len(args.servers) != len(args.xml_servers):
         print(
             'Number of servers arguments should be equal to the number of xmls provided.')
         sys.exit(1)
 
-    ds_procs = []
-    for i in range(0, len(args.servers)):
-        server_cmd = []
+    ds_procs = launch_discovery_server_processes(args.servers, args.xml_servers)
 
-        if not os.path.isfile(args.servers[i]):
-            print(f'Discovery server executable file does not exists: {args.servers[i]}')
-            sys.exit(1)
+    sub_procs, pub_procs = launch_client_processes(n_clients, pub_command, sub_command)
 
-        if not os.access(args.servers[i], os.X_OK):
-            print(
-                'Discovery server executable does not have execution permissions:'
-                f'{args.servers[i]}')
-            sys.exit(1)
+    terminate_ok = True
 
-        server_cmd.append(args.servers[i])
-        server_cmd.extend(['--xml-file', args.xml_servers[i]])
-        server_cmd.extend(['--server-id', str(i)])
+    if relaunch_clients:
+        time.sleep(3)
 
-        ds_proc = subprocess.Popen(server_cmd)
-        print(
-            'Running Discovery Server - commmand:  ',
-            ' '.join(map(str, server_cmd)))
+        cleanup_clients(pub_procs, sub_procs)
+        sub_procs, pub_procs = launch_client_processes(n_clients, pub_command, sub_command)
 
-        ds_procs.append(ds_proc)
+        time.sleep(3)
 
-    sub_proc = subprocess.Popen(sub_command)
-    print(
-            f'Running Subscriber - commmand:  ',
-            ' '.join(map(str, sub_command)))
-
-    pub_proc = subprocess.Popen(pub_command)
-    print(
-        'Running Publisher - commmand:  ',
-        ' '.join(map(str, pub_command)))
-
-    try:
-        outs, errs = sub_proc.communicate(timeout=15)
-    except subprocess.TimeoutExpired:
-        print('Subscriber process timed out, terminating...')
-        sub_proc.kill()
-        pub_proc.kill()
-        [ds_proc.kill() for ds_proc in ds_procs]
+    if args.validation_method == 'server':
+        # Check If discovery servers are still running
+        for ds_proc in ds_procs:
+            retcode = ds_proc.poll()
+            if retcode is not None and retcode is not 0:
+                print('Discovery Server process dead, terminating...')
+                terminate_ok = False
+    else:
         try:
-            sys.exit(os.EX_SOFTWARE)
-        except AttributeError:
-            sys.exit(1)
+            for sub_proc in sub_procs:
+                outs, errs = sub_proc.communicate(timeout=15)
 
+            if args.exit_on_disposal_received:
+                for pub_proc in pub_procs:
+                    outs, errs = pub_proc.communicate(timeout=5)
 
-    pub_proc.kill()
-    ds_proc.kill()
-    [ds_proc.kill() for ds_proc in ds_procs]
-    try:
-        sys.exit(os.EX_OK)
-    except AttributeError:
-        sys.exit(0)
+        except subprocess.TimeoutExpired:
+            print('Target process timed out, terminating...')
+            terminate_ok = False
 
+    cleanup(pub_procs, sub_procs, ds_procs)
+    terminate(terminate_ok)
 
 if __name__ == '__main__':
 
     # Parse arguments
     args = ParseOptions()
 
-    run(args.args)
+    run(args.args)