Validate that the port sizes are less than the MTU (#196)

* Validate that the port sizes are less than the MTU

* Choose CI runner version explicitly

* Add docs on how the max payload size must match the protobuf description

* Format

Author: erlingrj

include/reactor-uc/federated.h

@@ -11,7 +11,7 @@ typedef struct FederatedInputConnection FederatedInputConnection;
 typedef struct NetworkChannel NetworkChannel;
 
 // returns how many bytes of the buffer were used by the serialized string
-typedef size_t (*serialize_hook)(const void *user_struct, size_t user_struct_size, unsigned char *msg_buffer);
+typedef ssize_t (*serialize_hook)(const void *user_struct, size_t user_struct_size, unsigned char *msg_buffer);
 
 // returns if the deserialization was successful
 typedef lf_ret_t (*deserialize_hook)(void *user_struct, const unsigned char *msg_buffer, size_t msg_size);

include/reactor-uc/serialization.h

@@ -3,11 +3,18 @@
 
 #include "proto/message.pb.h"
 #include "reactor-uc/error.h"
+#include <stddef.h>
+
+// The maximum size of a serialized payload.
+// NOTE: This MUST match the max size of the payload in the protobuf message definition.
+#ifndef SERIALIZATION_MAX_PAYLOAD_SIZE
+#define SERIALIZATION_MAX_PAYLOAD_SIZE 832
+#endif
 
 int serialize_to_protobuf(const FederateMessage *message, unsigned char *buffer, size_t buffer_size);
 int deserialize_from_protobuf(FederateMessage *message, const unsigned char *buffer, size_t buffer_size);
 
 lf_ret_t deserialize_payload_default(void *user_struct, const unsigned char *msg_buf, size_t msg_size);
 
-size_t serialize_payload_default(const void *user_struct, size_t user_struct_size, unsigned char *msg_buf);
+ssize_t serialize_payload_default(const void *user_struct, size_t user_struct_size, unsigned char *msg_buf);
 #endif // REACTOR_UC_SERIALIZATION_H

src/federated.c

@@ -2,6 +2,7 @@
 #include "reactor-uc/environment.h"
 #include "reactor-uc/logging.h"
 #include "reactor-uc/platform.h"
+#include "reactor-uc/serialization.h"
 
 // TODO: Refactor so this function is available
 void LogicalConnection_trigger_downstreams(Connection *self, const void *value, size_t value_size);
@@ -91,14 +92,18 @@ void FederatedOutputConnection_cleanup(Trigger *trigger) {
     tagged_msg->tag.microstep = sched->current_tag(sched).microstep;
 
     assert(self->bundle->serialize_hooks[self->conn_id]);
-    size_t msg_size = (*self->bundle->serialize_hooks[self->conn_id])(self->staged_payload_ptr, self->payload_pool.size,
-                                                                      tagged_msg->payload.bytes);
-    tagged_msg->payload.size = msg_size;
-
-    LF_DEBUG(FED, "FedOutConn %p sending tagged message with tag=%" PRId64 ":%" PRIu32, trigger, tagged_msg->tag.time,
-             tagged_msg->tag.microstep);
-    if (channel->send_blocking(channel, &msg) != LF_OK) {
-      LF_ERR(FED, "FedOutConn %p failed to send message", trigger);
+    ssize_t msg_size = (*self->bundle->serialize_hooks[self->conn_id])(
+        self->staged_payload_ptr, self->payload_pool.size, tagged_msg->payload.bytes);
+    if (msg_size < 0) {
+      LF_ERR(FED, "Failed to serialize payload for federated output connection %p", trigger);
+    } else {
+      tagged_msg->payload.size = msg_size;
+
+      LF_DEBUG(FED, "FedOutConn %p sending tagged message with tag=%" PRId64 ":%" PRIu32, trigger, tagged_msg->tag.time,
+               tagged_msg->tag.microstep);
+      if (channel->send_blocking(channel, &msg) != LF_OK) {
+        LF_ERR(FED, "FedOutConn %p failed to send message", trigger);
+      }
     }
   } else {
     LF_WARN(FED, "FedOutConn %p not connected. Dropping staged message", trigger);
@@ -321,10 +326,12 @@ void FederatedConnectionBundle_validate(FederatedConnectionBundle *bundle) {
     validate(bundle->inputs[i]);
     validate(bundle->deserialize_hooks[i]);
     validate(bundle->inputs[i]->super.super.parent);
+    validate(bundle->inputs[i]->super.super.payload_pool->size < SERIALIZATION_MAX_PAYLOAD_SIZE);
   }
   for (size_t i = 0; i < bundle->outputs_size; i++) {
     validate(bundle->outputs[i]);
     validate(bundle->serialize_hooks[i]);
     validate(bundle->outputs[i]->super.super.parent);
+    validate(bundle->outputs[i]->super.super.payload_pool->size < SERIALIZATION_MAX_PAYLOAD_SIZE);
   }
 }

src/serialization.c

@@ -31,11 +31,17 @@ int deserialize_from_protobuf(FederateMessage *message, const unsigned char *buf
 }
 
 lf_ret_t deserialize_payload_default(void *user_struct, const unsigned char *msg_buf, size_t msg_size) {
-  memcpy(user_struct, msg_buf, MIN(msg_size, 832)); // TODO: 832 is a magic number
+  if (msg_size > SERIALIZATION_MAX_PAYLOAD_SIZE) {
+    return LF_ERR;
+  }
+  memcpy(user_struct, msg_buf, msg_size);
   return LF_OK;
 }
 
-size_t serialize_payload_default(const void *user_struct, size_t user_struct_size, unsigned char *msg_buf) {
-  memcpy(msg_buf, user_struct, MIN(user_struct_size, 832)); // TODO: 832 is a magic number
-  return MIN(user_struct_size, 832);                        // TODO: 832 is a magic number
+ssize_t serialize_payload_default(const void *user_struct, size_t user_struct_size, unsigned char *msg_buf) {
+  if (user_struct_size > SERIALIZATION_MAX_PAYLOAD_SIZE) {
+    return -1;
+  }
+  memcpy(msg_buf, user_struct, user_struct_size);
+  return user_struct_size;
 }