Draft: Apply pluggable security for federated execution.

core/federated/RTI/main.c

@@ -115,6 +115,8 @@ void usage(int argc, const char* argv[]) {
   lf_print("  -a, --auth Turn on HMAC authentication options.\n");
   lf_print("  -t, --tracing Turn on tracing.\n");
   lf_print("  -d, --disable_dnet Turn off the use of DNET signals.\n");
+  lf_print("  -sst, --sst SST config path for RTI.\n");
+  lf_print("  -tls, --tls <cert_path> <key_path>   TLS certificate and private key paths.\n");
 
   lf_print("Command given:");
   for (int i = 0; i < argc; i++) {
@@ -220,7 +222,7 @@ int process_args(int argc, const char* argv[]) {
       rti.base.number_of_scheduling_nodes = (int32_t)num_federates; // FIXME: Loses numbers on 64-bit machines
       lf_print_info("RTI: Number of federates: %d", rti.base.number_of_scheduling_nodes);
     } else if (strcmp(argv[i], "-p") == 0 || strcmp(argv[i], "--port") == 0) {
-#ifdef COMM_TYPE_TCP
+#if defined(COMM_TYPE_TCP) || defined(COMM_TYPE_SST) || defined(COMM_TYPE_TLS)
       if (argc < i + 2) {
         lf_print_error("--port needs a short unsigned integer argument ( > 0 and < %d).", UINT16_MAX);
         usage(argc, argv);
@@ -252,6 +254,42 @@ int process_args(int argc, const char* argv[]) {
       return 0;
 #endif
       rti.authentication_enabled = true;
+    } else if (strcmp(argv[i], "-sst") == 0 || strcmp(argv[i], "--sst") == 0) {
+#ifndef COMM_TYPE_SST
+      lf_print_error("--sst requires the RTI to be built with the --DCOMM_TYPE=SST option.");
+      usage(argc, argv);
+      return 0;
+#else
+      i++;
+      lf_set_sst_config_path(argv[i]);
+#endif
+    } else if (strcmp(argv[i], "-tls") == 0 || strcmp(argv[i], "--tls") == 0) {
+#ifndef COMM_TYPE_TLS
+      lf_print_error("--tls requires the RTI to be built with the -DCOMM_TYPE=TLS option.");
+      usage(argc, argv);
+      return 0;
+#else
+      // Need two arguments: cert path and key path
+      if (argc < i + 3) {
+        lf_print_error("--tls needs two arguments: <certificate_path> <private_key_path>.");
+        usage(argc, argv);
+        return 0;
+      }
+      const char* cert_path = argv[i + 1];
+      const char* key_path = argv[i + 2];
+
+      // Optional: basic sanity check (avoid empty strings)
+      if (cert_path[0] == '\0' || key_path[0] == '\0') {
+        lf_print_error("--tls certificate_path and private_key_path must be non-empty.");
+        usage(argc, argv);
+        return 0;
+      }
+
+      lf_set_tls_configuration(cert_path, key_path);
+      lf_print_debug("RTI: TLS cert path: %s", cert_path);
+      lf_print_debug("RTI: TLS key path : %s", key_path);
+      i += 2;
+#endif
     } else if (strcmp(argv[i], "-t") == 0 || strcmp(argv[i], "--tracing") == 0) {
       rti.base.tracing_enabled = true;
     } else if (strcmp(argv[i], "-d") == 0 || strcmp(argv[i], "--dnet_disabled") == 0) {

core/federated/RTI/rti_remote.c

@@ -636,23 +636,20 @@ void handle_address_query(uint16_t fed_id) {
 
   int32_t server_port;
   uint32_t* ip_address;
-  char* server_host_name;
+  uint32_t temp = 0;
 
   LF_MUTEX_LOCK(&rti_mutex);
   // Check if the RTI has initialized the remote federate's network abstraction.
   if (remote_fed->net == NULL) {
     // RTI has not set up the remote federate. Respond with -1 to indicate an unknown port number.
     server_port = -1;
-    uint32_t temp = 0;
     ip_address = &temp;
-    server_host_name = "localhost";
   } else {
     // The network abstraction is initialized, but the RTI might still not know the port number. This can happen if the
     // RTI has not yet received a MSG_TYPE_ADDRESS_ADVERTISEMENT message from the remote federate. In such cases, the
     // returned port number might still be -1.
-    server_port = ((socket_priv_t*)remote_fed->net)->server_port;
-    ip_address = (uint32_t*)&((socket_priv_t*)remote_fed->net)->server_ip_addr;
-    server_host_name = ((socket_priv_t*)remote_fed->net)->server_hostname;
+    server_port = get_server_port(remote_fed->net);
+    ip_address = (uint32_t*)get_ip_addr(remote_fed->net);
   }
 
   encode_int32(server_port, (unsigned char*)&buffer[1]);
@@ -669,8 +666,7 @@ void handle_address_query(uint16_t fed_id) {
     tracepoint_rti_to_federate(send_ADR_QR_REP, fed_id, NULL);
   }
 
-  LF_PRINT_DEBUG("Replied to address query from federate %d with address %s:%d.", fed_id, server_host_name,
-                 server_port);
+  LF_PRINT_DEBUG("Replied to address query from federate %d", fed_id);
 }
 
 void handle_address_ad(uint16_t federate_id) {
@@ -687,7 +683,8 @@ void handle_address_ad(uint16_t federate_id) {
   assert(server_port < 65536);
 
   LF_MUTEX_LOCK(&rti_mutex);
-  ((socket_priv_t*)fed->net)->server_port = server_port;
+  // (((sst_priv_t*)fed->net)->socket_priv)->server_port = server_port;
+  set_server_port(fed->net, server_port);
   LF_MUTEX_UNLOCK(&rti_mutex);
 
   LF_PRINT_LOG("Received address advertisement with port %d from federate %d.", server_port, federate_id);
@@ -1057,7 +1054,6 @@ void send_reject(net_abstraction_t net_abs, unsigned char error_code) {
   }
   // Close the network abstraction without reading until EOF.
   shutdown_net(net_abs, false);
-  net_abs = NULL;
   LF_MUTEX_UNLOCK(&rti_mutex);
 }
 
@@ -1325,7 +1321,7 @@ static int receive_udp_message_and_set_up_clock_sync(net_abstraction_t fed_net,
           // Initialize the UDP_addr field of the federate struct
           fed->UDP_addr.sin_family = AF_INET;
           fed->UDP_addr.sin_port = htons(federate_UDP_port_number);
-          fed->UDP_addr.sin_addr = ((socket_priv_t*)fed_net)->server_ip_addr;
+          fed->UDP_addr.sin_addr = *get_ip_addr(fed_net);
         }
       } else {
         // Disable clock sync after initial round.
@@ -1427,7 +1423,6 @@ void lf_connect_to_federates(net_abstraction_t rti_net) {
         lf_print_warning("RTI failed to authenticate the incoming federate.");
         // Close the network abstraction without reading until EOF.
         shutdown_net(fed_net, false);
-        fed_net = NULL;
         // Ignore the federate that failed authentication.
         i--;
         continue;
@@ -1496,7 +1491,6 @@ void* respond_to_erroneous_connections(void* nothing) {
     }
     // Close the network abstraction without reading until EOF.
     shutdown_net(fed_net, false);
-    fed_net = NULL;
   }
   return NULL;
 }
@@ -1513,7 +1507,7 @@ int start_rti_server() {
   // Initialize RTI's network abstraction.
   rti_remote->rti_net = initialize_net();
   // Set the user specified port to the network abstraction.
-  ((socket_priv_t*)rti_remote->rti_net)->user_specified_port = rti_remote->user_specified_port;
+  set_my_port(rti_remote->rti_net, rti_remote->user_specified_port);
   // Create the server
   if (create_server(rti_remote->rti_net)) {
     lf_print_error_system_failure("RTI failed to create TCP server: %s.", strerror(errno));

core/federated/RTI/rti_remote.h

@@ -271,7 +271,7 @@ void handle_address_query(uint16_t fed_id);
  * byte. The RTI will keep a record of this number in the .server_port
  * field of the _RTI.federates[federate_id] array of structs.
  *
- * The server_hostname and server_ip_addr fields are assigned
+ * The server_ip_addr field is assigned
  * in lf_connect_to_federates() upon accepting the socket
  * from the remote federate.
  *

core/federated/federate.c

@@ -641,11 +641,16 @@ static int handle_tagged_message(net_abstraction_t net, int fed_id) {
                      "    Discarding message and closing the network connection.",
                      env->current_tag.time - start_time, env->current_tag.microstep, intended_tag.time - start_time,
                      intended_tag.microstep);
-      // Free the allocated memory before returning
-      _lf_done_using(message_token);
-      // Close network abstraction, reading any incoming data and discarding it.
-      shutdown_net(_fed.net_for_inbound_p2p_connections[fed_id], false);
-      _fed.net_for_inbound_p2p_connections[fed_id] = NULL;
+      // The token was freshly allocated by _lf_new_token with ref_count 0 and was never
+      // scheduled, so _lf_done_using would incorrectly treat it as already freed.
+      // Use _lf_free_token directly, which handles ref_count == 0 correctly.
+      _lf_free_token(message_token);
+#ifdef FEDERATED_DECENTRALIZED
+      _lf_decrement_tag_barrier_locked(env);
+#endif
+      // Close the connection to unblock the listener, but do not free the memory;
+      // lf_terminate_execution will free it after joining the listener thread.
+      close_net(net, false);
       LF_MUTEX_UNLOCK(&env->mutex);
       return -1;
     } else {
@@ -723,7 +728,6 @@ static int handle_port_absent_message(net_abstraction_t net, int fed_id) {
  * network abstraction in _fed.net_for_inbound_p2p_connections
  * to -1 and returns, terminating the thread.
  * @param _args The remote federate ID (cast to void*).
- * @param fed_id_ptr A pointer to a uint16_t containing federate ID being listened to.
  *  This procedure frees the memory pointed to before returning.
  */
 static void* listen_to_federates(void* _args) {
@@ -1508,23 +1512,21 @@ static void* listen_to_rti_net(void* args) {
   // Listen for messages from the federate.
   while (!_lf_termination_executed) {
     // Check whether the RTI network abstraction is still valid.
-    if (_fed.net_to_RTI == NULL) {
-      lf_print_warning("network abstraction to the RTI unexpectedly closed.");
+    if (_fed.net_to_RTI == NULL || !is_net_open(_fed.net_to_RTI)) {
+      lf_print_warning("network connection to the RTI unexpectedly closed.");
       return NULL;
     }
     // Read one byte to get the message type.
     // This will exit if the read fails.
     int read_failed = read_from_net(_fed.net_to_RTI, 1, buffer);
     if (read_failed < 0) {
       lf_print_error("Connection to the RTI was closed by the RTI with an error. Considering this a soft error.");
-      shutdown_net(_fed.net_to_RTI, false);
-      _fed.net_to_RTI = NULL;
+      close_net(_fed.net_to_RTI, false);
       return NULL;
     } else if (read_failed > 0) {
       // EOF received.
       lf_print_info("Connection to the RTI closed with an EOF.");
-      shutdown_net(_fed.net_to_RTI, false);
-      _fed.net_to_RTI = NULL;
+      close_net(_fed.net_to_RTI, false);
       return NULL;
     }
     switch (buffer[0]) {
@@ -1647,22 +1649,17 @@ void lf_terminate_execution(environment_t* env) {
   }
 
   LF_PRINT_DEBUG("Closing incoming P2P network abstractions.");
-  // Close any incoming P2P network abstractions that are still open.
+  // Close connections to unblock any listener threads that are blocking on reads,
+  // but do NOT free the memory yet because listener threads hold local pointers.
   for (int i = 0; i < NUMBER_OF_FEDERATES; i++) {
-    shutdown_net(_fed.net_for_inbound_p2p_connections[i], false);
-    // Ignore errors. Mark the network abstraction closed.
-    _fed.net_for_inbound_p2p_connections[i] = NULL;
+    close_net(_fed.net_for_inbound_p2p_connections[i], false);
   }
 
   // Check for all outgoing physical connections in
   // _fed.net_for_outbound_p2p_connections and
   // if the network abstraction ID is not NULL, the connection is still open.
   // Send an EOF by closing the network abstraction here.
   for (int i = 0; i < NUMBER_OF_FEDERATES; i++) {
-
-    // Close outbound connections, in case they have not closed themselves.
-    // This will result in EOF being sent to the remote federate, except for
-    // abnormal termination, in which case it will just close the network abstraction.
     close_outbound_net(i);
   }
 
@@ -1681,6 +1678,14 @@ void lf_terminate_execution(environment_t* env) {
   // Wait for the thread listening for messages from the RTI to close.
   lf_thread_join(_fed.RTI_net_listener, NULL);
 
+  // All listener threads have now exited. Safe to free network abstraction memory.
+  for (int i = 0; i < NUMBER_OF_FEDERATES; i++) {
+    free_net(_fed.net_for_inbound_p2p_connections[i]);
+    _fed.net_for_inbound_p2p_connections[i] = NULL;
+  }
+  free_net(_fed.net_to_RTI);
+  _fed.net_to_RTI = NULL;
+
   // For abnormal termination, there is no need to free memory.
   if (_lf_normal_termination) {
     LF_PRINT_DEBUG("Freeing memory occupied by the federate.");
@@ -1753,13 +1758,24 @@ void lf_connect_to_federate(uint16_t remote_federate_id) {
   assert(port > 0);
   uint16_t uport = (uint16_t)port;
 
-  char hostname[INET_ADDRSTRLEN];
-  inet_ntop(AF_INET, &host_ip_addr, hostname, INET_ADDRSTRLEN);
-
-  socket_connection_params_t params;
+#ifdef COMM_TYPE_TCP
+  socket_connection_params_t params = {0};
   params.type = TCP;
   params.port = uport;
-  params.server_hostname = hostname;
+  params.server_ip_addr = &host_ip_addr;
+#elif defined(COMM_TYPE_SST)
+  sst_connection_params_t params = {0};
+  params.socket_params.type = TCP;
+  params.socket_params.port = uport;
+  params.socket_params.server_ip_addr = &host_ip_addr;
+  params.target = 1;
+#elif defined(COMM_TYPE_TLS)
+  tls_connection_params_t params = {0};
+  params.socket_params.type = TCP;
+  params.socket_params.port = uport;
+  params.socket_params.server_ip_addr = &host_ip_addr;
+#endif
+
   net_abstraction_t net = connect_to_net((net_params_t)&params);
   if (net == NULL) {
     lf_print_error_and_exit("Failed to connect to federate.");
@@ -1837,10 +1853,24 @@ void lf_connect_to_rti(const char* hostname, int port) {
   hostname = federation_metadata.rti_host ? federation_metadata.rti_host : hostname;
   port = federation_metadata.rti_port >= 0 ? federation_metadata.rti_port : port;
 
-  socket_connection_params_t params;
+#ifdef COMM_TYPE_TCP
+  socket_connection_params_t params = {0};
   params.type = TCP;
   params.port = port;
   params.server_hostname = hostname;
+#elif defined(COMM_TYPE_SST)
+  sst_connection_params_t params = {0};
+  params.socket_params.type = TCP;
+  params.socket_params.port = port;
+  params.socket_params.server_hostname = hostname;
+  params.target = 0;
+#elif defined(COMM_TYPE_TLS)
+  tls_connection_params_t params = {0};
+  params.socket_params.type = TCP;
+  params.socket_params.port = port;
+  params.socket_params.server_hostname = hostname;
+#endif
+
   net_abstraction_t net = connect_to_net((net_params_t)&params);
   if (net == NULL) {
     lf_print_error_and_exit("Failed to connect to RTI.");
@@ -1949,14 +1979,13 @@ void lf_create_server(int specified_port) {
   assert(specified_port <= UINT16_MAX && specified_port >= 0);
 
   net_abstraction_t server_net = initialize_net();
-  ((socket_priv_t*)server_net)->port = (uint16_t)specified_port;
-
+  set_my_port(server_net, specified_port);
   if (create_server(server_net)) {
     lf_print_error_system_failure("Failed to create server: %s.", strerror(errno));
   };
   _fed.server_net = server_net;
   // Get the final server port to send to the RTI on an MSG_TYPE_ADDRESS_ADVERTISEMENT message.
-  int32_t server_port = ((socket_priv_t*)server_net)->port;
+  int32_t server_port = get_my_port(server_net);
 
   LF_PRINT_LOG("Server for communicating with other federates started using port %d.", server_port);
 
@@ -2007,13 +2036,14 @@ void* lf_handle_p2p_connections_from_federates(void* env_arg) {
   _fed.inbound_net_listeners = (lf_thread_t*)calloc(_fed.number_of_inbound_p2p_connections, sizeof(lf_thread_t));
   while (received_federates < _fed.number_of_inbound_p2p_connections && !_lf_termination_executed) {
     if (rti_failed()) {
-      break;
+      return NULL;
     }
     // Wait for an incoming connection request.
     net_abstraction_t net = accept_net(_fed.server_net);
     if (net == NULL) {
       lf_print_warning("Federate failed to accept the network abstraction.");
-      return NULL;
+      lf_sleep(CONNECT_RETRY_INTERVAL);
+      continue;
     }
     LF_PRINT_LOG("Accepted new connection from remote federate.");
 
@@ -2034,7 +2064,6 @@ void* lf_handle_p2p_connections_from_federates(void* env_arg) {
         write_to_net(net, 2, response);
       }
       shutdown_net(net, false);
-      net = NULL;
       continue;
     }
 
@@ -2055,7 +2084,6 @@ void* lf_handle_p2p_connections_from_federates(void* env_arg) {
         write_to_net(net, 2, response);
       }
       shutdown_net(net, false);
-      net = NULL;
       continue;
     }