Merge pull request #70 from insa-unyte/feature/ipv6-support

Feature/ipv6 support

Author: ahuangfeng

README.md

@@ -28,7 +28,7 @@ Simple example of usage :
 #include <unyte-udp-notif/unyte_udp_utils.h>
 #include <unyte-udp-notif/unyte_udp_queue.h>
 
-#define PORT 10001
+#define PORT "10001"
 #define ADDR "192.168.0.17"
 
 int main()
@@ -61,9 +61,8 @@ int main()
     printf("unyte_udp_get_message_length: %u\n", unyte_udp_get_message_length(seg));
     printf("unyte_udp_get_generator_id: %u\n", unyte_udp_get_generator_id(seg));
     printf("unyte_udp_get_message_id: %u\n", unyte_udp_get_message_id(seg));
-    printf("unyte_udp_get_src_port: %u\n", unyte_udp_get_src_port(seg));
-    printf("unyte_udp_get_src_addr: %u\n", unyte_udp_get_src_addr(seg));
-    printf("unyte_udp_get_dest_addr: %u\n", unyte_udp_get_dest_addr(seg));
+    printf("unyte_udp_get_src[family]: %u\n", unyte_udp_get_src(seg)->ss_family);               // AF_INET for IPv4 or AF_INET6 for IPv6
+    printf("unyte_udp_get_dest_addr[family]: %u\n", unyte_udp_get_dest_addr(seg)->ss_family);   // AF_INET for IPv4 or AF_INET6 for IPv6
     printf("unyte_udp_get_payload: %s\n", unyte_udp_get_payload(seg));
     printf("unyte_udp_get_payload_length: %u\n", unyte_udp_get_payload_length(seg));
 
@@ -102,9 +101,8 @@ typedef struct unyte_segment_with_metadata
 - `uint16_t unyte_udp_get_message_length(unyte_seg_met_t *message);` : total length of the message within one UDP datagram, measured in octets, including the message header
 - `uint32_t unyte_udp_get_generator_id(unyte_seg_met_t *message);` : observation domain id of the message
 - `uint32_t unyte_udp_get_message_id(unyte_seg_met_t *message);` : message id of the message
-- `uint16_t unyte_udp_get_src_port(unyte_seg_met_t *message);` : source port of the message
-- `uint32_t unyte_udp_get_src_addr(unyte_seg_met_t *message);` : source address of the message
-- `uint32_t unyte_udp_get_dest_addr(unyte_seg_met_t *message);` : collector address
+- `struct sockaddr_storage * unyte_udp_get_src(unyte_seg_met_t *message);` : source IP and port of the message. Could be IPv4 or IPv6.
+- `struct sockaddr_storage * unyte_udp_get_dest_addr(unyte_seg_met_t *message);` : collector address. Could be IPv4 or IPv6.
 - `char *unyte_udp_get_payload(unyte_seg_met_t *message);` : payload buffer
 - `uint16_t unyte_udp_get_payload_length(unyte_seg_met_t *message);` : payload length
 
@@ -115,7 +113,7 @@ To activate this thread, you must initiate the monitoring thread queue size (`mo
 typedef struct
 {
   char *address;
-  uint16_t port;
+  char *port;
   ...
   uint monitoring_queue_size;   // monitoring queue size if wanted to activate the monitoring thread. Default: 0. Recommended: 500.
   uint monitoring_delay;        // monitoring queue frequence in seconds. Default: 5 seconds
@@ -163,7 +161,7 @@ Simple usage of the sender :
 #include <unyte-udp-notif/unyte_sender.h>
 #include <unyte-udp-notif/unyte_udp_utils.h>
 
-#define PORT 10001
+#define PORT "10001"
 #define ADDR "192.168.0.17"
 #define MTU 1500
 

examples/client_monitoring.c

@@ -45,12 +45,12 @@ int main(int argc, char *argv[])
   // Initialize collector options
   unyte_udp_options_t options = {0};
   options.address = argv[1];
-  options.port = atoi(argv[2]);
+  options.port = argv[2];
   options.recvmmsg_vlen = USED_VLEN;
   options.monitoring_delay = 2;
   options.monitoring_queue_size = 500;
 
-  printf("Listening on %s:%d\n", options.address, options.port);
+  printf("Listening on %s:%s\n", options.address, options.port);
 
   /* Initialize collector */
   unyte_udp_collector_t *collector = unyte_udp_start_collector(&options);

examples/client_sample.c

@@ -5,14 +5,15 @@
 #include <string.h>
 #include <signal.h>
 #include <unistd.h>
+#include <arpa/inet.h>
 
 #include "../src/hexdump.h"
 #include "../src/unyte_udp_collector.h"
 #include "../src/unyte_udp_utils.h"
 #include "../src/unyte_udp_queue.h"
 
-#define USED_VLEN 10
-#define MAX_TO_RECEIVE 200
+#define USED_VLEN 1
+#define MAX_TO_RECEIVE 20
 
 int main(int argc, char *argv[])
 {
@@ -27,8 +28,8 @@ int main(int argc, char *argv[])
   unyte_udp_options_t options = {0};
   options.recvmmsg_vlen = USED_VLEN;
   options.address = argv[1];
-  options.port = atoi(argv[2]);
-  printf("Listening on %s:%d\n", options.address, options.port);
+  options.port = argv[2];
+  printf("Listening on %s:%s\n", options.address, options.port);
 
   /* Initialize collector */
   unyte_udp_collector_t *collector = unyte_udp_start_collector(&options);
@@ -53,9 +54,24 @@ int main(int argc, char *argv[])
     // printf("unyte_udp_get_message_length: %u\n", unyte_udp_get_message_length(seg));
     // printf("unyte_udp_get_generator_id: %u\n", unyte_udp_get_generator_id(seg));
     // printf("unyte_udp_get_message_id: %u\n", unyte_udp_get_message_id(seg));
-    // printf("unyte_udp_get_src_port: %u\n", unyte_udp_get_src_port(seg));
-    // printf("unyte_udp_get_src_addr: %u\n", unyte_udp_get_src_addr(seg));
-    // printf("unyte_udp_get_dest_addr: %u\n", unyte_udp_get_dest_addr(seg));
+    // printf("unyte_udp_get_src[family]: %u\n", unyte_udp_get_src(seg)->ss_family);
+    // printf("unyte_udp_get_dest_addr[family]: %u\n", unyte_udp_get_dest_addr(seg)->ss_family);
+    // char ip_canonical[100];
+    // if (unyte_udp_get_src(seg)->ss_family == AF_INET) {
+    //   printf("src IPv4: %s\n", inet_ntop(unyte_udp_get_src(seg)->ss_family, &((struct sockaddr_in*)unyte_udp_get_src(seg))->sin_addr.s_addr, ip_canonical, sizeof ip_canonical));
+    //   printf("src port: %u\n", ntohs(((struct sockaddr_in*)unyte_udp_get_src(seg))->sin_port));
+    // } else {
+    //   printf("src IPv6: %s\n", inet_ntop(unyte_udp_get_src(seg)->ss_family, &((struct sockaddr_in6*)unyte_udp_get_src(seg))->sin6_addr.s6_addr, ip_canonical, sizeof ip_canonical));
+    //   printf("src port: %u\n", ntohs(((struct sockaddr_in6*)unyte_udp_get_src(seg))->sin6_port));
+    // }
+    // char ip_dest_canonical[100];
+    // if (unyte_udp_get_src(seg)->ss_family == AF_INET) {
+    //   printf("dest IPv4: %s\n", inet_ntop(unyte_udp_get_dest_addr(seg)->ss_family, &((struct sockaddr_in*)unyte_udp_get_dest_addr(seg))->sin_addr.s_addr, ip_dest_canonical, sizeof ip_dest_canonical));
+    //   printf("dest port: %u\n", ntohs(((struct sockaddr_in*)unyte_udp_get_dest_addr(seg))->sin_port));
+    // } else {
+    //   printf("dest IPv6: %s\n", inet_ntop(unyte_udp_get_dest_addr(seg)->ss_family, &((struct sockaddr_in6*)unyte_udp_get_dest_addr(seg))->sin6_addr.s6_addr, ip_dest_canonical, sizeof ip_dest_canonical));
+    //   printf("dest port: %u\n", ntohs(((struct sockaddr_in6*)unyte_udp_get_dest_addr(seg))->sin6_port));
+    // }
     // printf("unyte_udp_get_payload: %s\n", unyte_udp_get_payload(seg));
     // printf("unyte_udp_get_payload_length: %u\n", unyte_udp_get_payload_length(seg));
 

src/listening_worker.c

@@ -59,6 +59,7 @@ void free_parsers(struct parse_worker *parsers, struct listener_thread_input *in
     free(messages[i].msg_hdr.msg_iov->iov_base);
     free(messages[i].msg_hdr.msg_iov);
     free(messages[i].msg_hdr.msg_name);
+    free(messages[i].msg_hdr.msg_control);
   }
   free(messages);
 }
@@ -185,6 +186,45 @@ int create_monitoring_thread(struct monitoring_worker *monitoring, unyte_udp_que
   return 0;
 }
 
+struct sockaddr_storage *get_dest_addr(struct msghdr *mh, unyte_udp_sock_t *sock)
+{
+  struct sockaddr_storage *addr = (struct sockaddr_storage *)malloc(sizeof(struct sockaddr_storage));
+
+  if (addr == NULL)
+  {
+    printf("Malloc failed\n");
+    return NULL;
+  }
+
+  memset(addr, 0, sizeof(struct sockaddr_storage));
+  struct in_pktinfo *in_pktinfo;
+  struct in6_pktinfo *in6_pktinfo;
+
+  for ( // iterate through all the control headers
+      struct cmsghdr *cmsg = CMSG_FIRSTHDR(mh);
+      cmsg != NULL;
+      cmsg = CMSG_NXTHDR(mh, cmsg))
+  {
+    if (cmsg->cmsg_level == IPPROTO_IP && cmsg->cmsg_type == IP_PKTINFO)
+    {
+      in_pktinfo = (struct in_pktinfo *)CMSG_DATA(cmsg);
+      ((struct sockaddr_in *)addr)->sin_family = AF_INET;
+      ((struct sockaddr_in *)addr)->sin_addr = in_pktinfo->ipi_addr;
+      ((struct sockaddr_in *)addr)->sin_port = ((struct sockaddr_in *)sock->addr)->sin_port;
+      break;
+    }
+    if (cmsg->cmsg_level == IPPROTO_IPV6 && cmsg->cmsg_type == IPV6_PKTINFO)
+    {
+      in6_pktinfo = (struct in6_pktinfo *)CMSG_DATA(cmsg);
+      ((struct sockaddr_in6 *)addr)->sin6_family = AF_INET6;
+      ((struct sockaddr_in6 *)addr)->sin6_addr = in6_pktinfo->ipi6_addr;
+      ((struct sockaddr_in6 *)addr)->sin6_port = ((struct sockaddr_in6 *)sock->addr)->sin6_port;
+      break;
+    }
+  }
+  return addr;
+}
+
 /**
  * Udp listener worker on PORT port.
  */
@@ -234,11 +274,14 @@ int listener(struct listener_thread_input *in)
     printf("Malloc failed \n");
     return -1;
   }
+  int cmbuf_len = 1024;
   // Init msg_iov first to reduce mallocs every iteration of while
   for (uint16_t i = 0; i < in->recvmmsg_vlen; i++)
   {
     messages[i].msg_hdr.msg_iov = (struct iovec *)malloc(sizeof(struct iovec));
     messages[i].msg_hdr.msg_iovlen = 1;
+    messages[i].msg_hdr.msg_control = (char *)malloc(cmbuf_len);
+    messages[i].msg_hdr.msg_controllen = cmbuf_len;
   }
   // FIXME: malloc failed
   // TODO: malloc array of msg_hdr.msg_name and assign addresss to every messages[i]
@@ -251,12 +294,10 @@ int listener(struct listener_thread_input *in)
     {
       messages[i].msg_hdr.msg_iov->iov_base = (char *)malloc(UDP_SIZE * sizeof(char));
       messages[i].msg_hdr.msg_iov->iov_len = UDP_SIZE;
-      messages[i].msg_hdr.msg_control = 0;
-      messages[i].msg_hdr.msg_controllen = 0;
-      messages[i].msg_hdr.msg_name = (struct sockaddr_in *)malloc(sizeof(struct sockaddr_in));
-      messages[i].msg_hdr.msg_namelen = sizeof(struct sockaddr_in);
+      memset(messages[i].msg_hdr.msg_control, 0, cmbuf_len);
+      messages[i].msg_hdr.msg_name = (struct sockaddr_storage *)malloc(sizeof(struct sockaddr_storage));
+      messages[i].msg_hdr.msg_namelen = sizeof(struct sockaddr_storage);
     }
-
     int read_count = recvmmsg(*in->conn->sockfd, messages, in->recvmmsg_vlen, 0, NULL);
     if (read_count == -1)
     {
@@ -273,7 +314,12 @@ int listener(struct listener_thread_input *in)
       // If msg_len == 0 -> message has 0 bytes -> we discard message and free the buffer
       if (messages[i].msg_len > 0)
       {
-        unyte_min_t *seg = minimal_parse(messages[i].msg_hdr.msg_iov->iov_base, ((struct sockaddr_in *)messages[i].msg_hdr.msg_name), in->conn->addr);
+        struct sockaddr_storage *dest_addr = get_dest_addr(&(messages[i].msg_hdr), in->conn);
+
+        if (dest_addr == NULL)
+          return -1;
+
+        unyte_min_t *seg = minimal_parse(messages[i].msg_hdr.msg_iov->iov_base, ((struct sockaddr_storage *)messages[i].msg_hdr.msg_name), dest_addr);
         if (seg == NULL)
         {
           printf("minimal_parse error\n");
@@ -303,11 +349,6 @@ int listener(struct listener_thread_input *in)
         free(messages[i].msg_hdr.msg_iov->iov_base);
       }
     }
-
-    for (uint16_t i = 0; i < in->recvmmsg_vlen; i++)
-    {
-      free(messages[i].msg_hdr.msg_name);
-    }
   }
 
   // Never called cause while(1)

src/listening_worker.h

@@ -18,7 +18,6 @@ struct listener_thread_input
 {
   unyte_udp_queue_t *output_queue;      /* The queue used to push the segments outside. */
   unyte_udp_queue_t *monitoring_queue;  /* The queue used to push monitoring stats of segments outside. */
-  uint16_t port;              /* The port to initialize the interface on. */
   unyte_udp_sock_t *conn;     /* Connection with addr, sockfd */
   uint16_t recvmmsg_vlen;     /* The recvmmsg buffer array size */
   uint nb_parsers;            /* Number of parsers instances to init */

src/parsing_worker.c

@@ -36,6 +36,17 @@ unyte_seg_met_t *create_assembled_msg(char *complete_msg, unyte_seg_met_t *src_p
     return NULL;
   }
 
+  parsed_msg->metadata->src = (struct sockaddr_storage *)malloc(sizeof(struct sockaddr_storage));
+  parsed_msg->metadata->dest = (struct sockaddr_storage *)malloc(sizeof(struct sockaddr_storage));
+
+  if (parsed_msg->metadata->src == NULL || parsed_msg->metadata->dest == NULL)
+  {
+    free(parsed_msg->metadata);
+    free(parsed_msg->header);
+    free(parsed_msg);
+    return NULL;
+  }
+
   copy_unyte_seg_met_headers(parsed_msg, src_parsed_segment);
 
   // Rewrite header length and message length

src/unyte_sender.c

@@ -4,31 +4,45 @@
 #include <unistd.h>
 #include <netinet/in.h>
 #include <arpa/inet.h>
-#include <linux/if.h>
 #include <string.h>
 #include "unyte_sender.h"
+#include <net/if.h>
 
 struct unyte_sender_socket *unyte_start_sender(unyte_sender_options_t *options)
 {
-  struct sockaddr_in *addr = (struct sockaddr_in *)malloc(sizeof(struct sockaddr_in));
+  unyte_IP_type_t ip_type = get_IP_type(options->address);
+  if (ip_type == unyte_UNKNOWN)
+  {
+    printf("Invalid IP address: %s\n", options->address);
+    exit(EXIT_FAILURE);
+  }
+
+  void *addr;
   struct unyte_sender_socket *sender_sk = (struct unyte_sender_socket *)malloc(sizeof(struct unyte_sender_socket));
+
+  if (ip_type == unyte_IPV4)
+    addr = (struct sockaddr_in *)malloc(sizeof(struct sockaddr_in));
+  else
+    addr = (struct sockaddr_in6 *)malloc(sizeof(struct sockaddr_in6));
+
   if (addr == NULL || sender_sk == NULL)
   {
     perror("Malloc failed.\n");
     exit(EXIT_FAILURE);
   }
 
-  int sockfd = socket(AF_INET, SOCK_DGRAM, 0);
+  int sockfd;
+  if (ip_type == unyte_IPV4)
+    sockfd = socket(AF_INET, SOCK_DGRAM, 0);
+  else
+    sockfd = socket(AF_INET6, SOCK_DGRAM, 0);
+
   if (sockfd < 0)
   {
     perror("socket()");
     exit(EXIT_FAILURE);
   }
 
-  addr->sin_family = AF_INET;
-  addr->sin_port = htons(options->port);
-  inet_pton(AF_INET, options->address, &addr->sin_addr);
-
   // Binding socket to an interface
   const char *interface = options->interface;
   if (options->interface != NULL && (strlen(interface) > 0))
@@ -57,10 +71,30 @@ struct unyte_sender_socket *unyte_start_sender(unyte_sender_options_t *options)
     exit(EXIT_FAILURE);
   }
 
+  int connect_ret = 0;
+  if (ip_type == unyte_IPV4)
+  {
+    memset(addr, 0, sizeof(struct sockaddr_in));
+    ((struct sockaddr_in *)addr)->sin_family = AF_INET;
+    ((struct sockaddr_in *)addr)->sin_port = htons(options->port);
+    inet_pton(AF_INET, options->address, &((struct sockaddr_in *)addr)->sin_addr);
+    connect_ret = connect(sockfd, (struct sockaddr *)addr, sizeof(struct sockaddr_in));
+  }
+  else
+  {
+    memset(addr, 0, sizeof(struct sockaddr_in6));
+    ((struct sockaddr_in6 *)addr)->sin6_family = AF_INET6;
+    ((struct sockaddr_in6 *)addr)->sin6_port = htons(options->port);
+    ((struct sockaddr_in6 *)addr)->sin6_scope_id = if_nametoindex("eth0"); //TODO: check all interfaces or bind to one interface
+    ((struct sockaddr_in6 *)addr)->sin6_flowinfo = 0;
+    inet_pton(AF_INET6, options->address, &((struct sockaddr_in6 *)addr)->sin6_addr);
+    connect_ret = connect(sockfd, (struct sockaddr *)addr, sizeof(struct sockaddr_in6));
+  }
+
   // connect socket to destination address
-  if (connect(sockfd, addr, sizeof(struct sockaddr_in)) == -1)
+  if (connect_ret == -1)
   {
-    perror("Bind failed");
+    perror("Connect failed");
     close(sockfd);
     exit(EXIT_FAILURE);
   }