Fix SecurityManager assertion in Secure DS (#5329)

* Refs #21696: Add regression test

Signed-off-by: Mario Dominguez <mariodominguez@eprosima.com>

* Refs #21696: Add fix

Signed-off-by: Mario Dominguez <mariodominguez@eprosima.com>

---------

Signed-off-by: Mario Dominguez <mariodominguez@eprosima.com>
(cherry picked from commit ce72a14)

src/cpp/rtps/security/SecurityManager.cpp

@@ -946,6 +946,10 @@ bool SecurityManager::on_process_handshake(
                 else
                 {
                     EPROSIMA_LOG_ERROR(SECURITY, "WriterHistory cannot add the CacheChange_t");
+                    // Return the handshake handle
+                    authentication_plugin_->return_handshake_handle(remote_participant_info->handshake_handle_,
+                            exception);
+                    remote_participant_info->handshake_handle_ = nullptr;
                     participant_stateless_message_writer_history_->release_change(change);
                 }
             }
@@ -957,6 +961,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

@@ -390,6 +390,28 @@ 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()
+
     endif()
 
 endif()

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,100 @@ 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].'
+        )
 
         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 +200,8 @@ def run(args):
     """
     pub_command = []
     sub_command = []
+    n_clients = 1
+    relaunch_clients = False
 
     script_dir = os.path.dirname(os.path.realpath(__file__))
 
@@ -156,71 +250,52 @@ def run(args):
         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)
-
-
-    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)
+            for sub_proc in sub_procs:
+                outs, errs = sub_proc.communicate(timeout=15)
+        except subprocess.TimeoutExpired:
+            print('Subscriber 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)