Live PRKI Origin Validation Annotation

- The BGPStream will be extended by Live PRKI Origin Validation
Annotation

- All details concerning the provided functions, annotation elements
and output format are described in issue #19 and
/pull/26 (update)

Author: salsh

configure.ac

@@ -262,20 +262,42 @@ AC_DEFINE_UNQUOTED([ED_PLUGIN_INIT_ALL_ENABLED], $ED_PLUGIN_INIT_ALL_ENABLED,
 
 AC_MSG_NOTICE([----------------------------------])
 
+# RTR configuration
 AC_MSG_CHECKING([whether the RTR library is available])
+AC_ARG_WITH([rtr],
+	[AS_HELP_STRING([--without-rtr],
+	  [do not compile with rtr support])],
+	  [],
+	  [with_rtr=yes])
+AM_CONDITIONAL([WITH_rtr], [test "x$with_rtr" != xno])
+if test x"$with_rtr" = xyes; then
 AC_LINK_IFELSE([AC_LANG_PROGRAM([[#include "rtrlib/rtrlib.h"]])],
       [AC_MSG_RESULT(yes)
-       AC_DEFINE(FOUND_RTR,,found_rtr)],
+       AC_DEFINE([WITH_RTR],[1],[Building with RTR support])],
        AC_MSG_RESULT(no)
   )
+else
+AC_MSG_RESULT([no])
+fi
 
+# SSH configuration
 AC_MSG_CHECKING([whether the RTR library is compiled with SSH])
+AC_ARG_WITH([ssh],
+	[AS_HELP_STRING([--without-ssh],
+	  [do not compile with ssh support])],
+	  [],
+	  [with_ssh=yes])
+AM_CONDITIONAL([WITH_ssh], [test "x$with_ssh" != xno])
+if test x"$with_ssh" = xyes; then
 AC_LINK_IFELSE([AC_LANG_PROGRAM([[#include "rtrlib/rtrlib.h"]],
                         [[struct tr_ssh_config config;]])],
       [AC_MSG_RESULT(yes)
-       AC_DEFINE(FOUND_SSH,,found_ssh)],
+       AC_DEFINE([WITH_SSH],[1],[Building with SSH support])],
        AC_MSG_RESULT(no)
   )
+else
+AC_MSG_RESULT([no])
+fi
 
 # we may want to come back later and add compile-time configuration for things
 # like datastructure providers, but for now it will all get compiled

lib/bgpstream.c

@@ -199,6 +199,11 @@ bgpstream_t *bgpstream_create() {
     bs = NULL;
     return NULL;
   }
+
+#ifdef WITH_RTR
+  bs->rtr_server_conf.active = 0;
+#endif
+
   /* memory for the bgpstream interface has been
    * allocated correctly */
   bs->status = BGPSTREAM_STATUS_ALLOCATED;
@@ -401,27 +406,27 @@ void bgpstream_set_live_mode(bgpstream_t *bs) {
   bgpstream_debug("BS: set_blocking stop");
 }
 
-#if defined(FOUND_RTR)
+#ifdef WITH_RTR
 /* Get the RTR-Socket & configuration
 */
-struct rtr_mgr_config *bgpstream_get_rtr_config()
+struct rtr_mgr_config *bgpstream_get_rtr_config(bgpstream_t *bs)
 {
-  return cfg_tr;
+  return bs->cfg_tr;
 }
 
 /* Set the RTR-Configuration
 */
-int bgpstream_set_rtr_config(char *host, char *port, char *ssh_user,
+int bgpstream_set_rtr_config(bgpstream_t *bs, char *host, char *port, char *ssh_user,
                              char *ssh_hostkey, char *ssh_privatekey,
-                             bool active)
+                             int active)
 {
-  rtr_server_conf.host = host;
-  rtr_server_conf.port = port;
-  rtr_server_conf.ssh_user = ssh_user;
-  rtr_server_conf.ssh_hostkey = ssh_hostkey;
-  rtr_server_conf.ssh_privatekey = ssh_privatekey;
-  rtr_server_conf.active = active;
 
+  bs->rtr_server_conf.host = host;
+  bs->rtr_server_conf.port = port;
+  bs->rtr_server_conf.ssh_user = ssh_user;
+  bs->rtr_server_conf.ssh_hostkey = ssh_hostkey;
+  bs->rtr_server_conf.ssh_privatekey = ssh_privatekey;
+  bs->rtr_server_conf.active = active;
   return 0;
 }
 #endif
@@ -433,12 +438,12 @@ int bgpstream_set_rtr_config(char *host, char *port, char *ssh_user,
 */
 int bgpstream_start(bgpstream_t *bs) {
   bgpstream_debug("BS: init start");
-#if defined(FOUND_RTR)
-  if (rtr_server_conf.active) {
-    cfg_tr = bgpstream_rtr_start_connection(
-        rtr_server_conf.host, rtr_server_conf.port, NULL, NULL, NULL,
-        rtr_server_conf.ssh_user, rtr_server_conf.ssh_hostkey,
-        rtr_server_conf.ssh_privatekey);
+#ifdef WITH_RTR
+  if (bs->rtr_server_conf.active) {
+    bs->cfg_tr = bgpstream_rtr_start_connection(
+        bs->rtr_server_conf.host, bs->rtr_server_conf.port, NULL, NULL, NULL,
+        bs->rtr_server_conf.ssh_user, bs->rtr_server_conf.ssh_hostkey,
+        bs->rtr_server_conf.ssh_privatekey);
   }
 #endif
   if(bs == NULL || (bs != NULL && bs->status != BGPSTREAM_STATUS_ALLOCATED)) {
@@ -525,10 +530,10 @@ int bgpstream_get_next_record(bgpstream_t *bs,
 /* turn off the bgpstream interface */
 void bgpstream_stop(bgpstream_t *bs) {
   bgpstream_debug("BS: close start");
-#if defined(FOUND_RTR)
-  if (rtr_server_conf.active) {
-    bgpstream_rtr_close_connection(cfg_tr);
-    rtr_server_conf.active = false;
+#ifdef WITH_RTR
+  if (bs->rtr_server_conf.active) {
+    bgpstream_rtr_close_connection(bs->cfg_tr);
+    bs->rtr_server_conf.active = false;
   }
 #endif
   if(bs == NULL || (bs != NULL && bs->status != BGPSTREAM_STATUS_ON)) {

lib/bgpstream.h

@@ -146,17 +146,7 @@ typedef struct struct_bgpstream_data_interface_option
 
 } bgpstream_data_interface_option_t;
 
-#if defined(FOUND_RTR)
-struct rtr_mgr_config *cfg_tr;
-
-struct rtr_server_configure {
-  char *host;
-  char *port;
-  char *ssh_user;
-  char *ssh_hostkey;
-  char *ssh_privatekey;
-  bool active;
-} rtr_server_conf;
+#ifdef WITH_RTR
 
 /** @} */
 
@@ -173,13 +163,18 @@ struct rtr_mgr_config *bgpstream_get_rtr_config();
 
 /** Set the configuration of the RTR-Socket-Manager
 *
-* @param host    the host of the cache server
-* @param port    the port of the cache server
-* @param active  whether the rtr-validation is enabled
+* @param bs             a pointer to a BGP Stream instance
+* @param host           the host of the cache server
+* @param port           the port of the cache server
+* @param ssh_user       the username for a SSH connection
+* @param ssh_hostkey    the hostkey for a SSH connection
+* @param ssh_privatekey the private key for a SSH connection
+* @param active         whether the rtr-validation is enabled
 */
-int bgpstream_set_rtr_config(char *host, char *port, char *ssh_user,
+int bgpstream_set_rtr_config(bgpstream_t *bs,
+                              char *host, char *port, char *ssh_user,
                              char *ssh_hostkey, char *ssh_privatekey,
-                             bool active);
+                             int active);
 #endif
 
 /** Create a new BGP Stream instance

lib/bgpstream_constants.h

@@ -30,6 +30,12 @@
 // dump name max length
 #define BGPSTREAM_DUMP_MAX_LEN 1024
 
+// RPKI validation result max length
+#define BGPSTREAM_RPKI_RST_MAX_LEN 2048
+
+// RPKI validation result max ROA entries
+#define BGPSTREAM_RPKI_MAX__ROA_ENT 16
+
 // parameters/attribute/filters max length
 #define BGPSTREAM_PAR_MAX_LEN 512
 

lib/bgpstream_elem.c

@@ -30,15 +30,17 @@
 
 #include "bgpdump_lib.h"
 #include "utils.h"
+#include "khash.h"
 
 #include "bgpstream.h"
 #include "bgpstream_utils.h"
 
 #include "bgpstream_debug.h"
 #include "bgpstream_record.h"
 
+#include "bgpstream_constants.h"
 #include "bgpstream_elem_int.h"
-#include "utils/bgpstream_utils_rtr.h"
+#include "bgpstream_utils_rtr.h"
 
 /* ==================== PROTECTED FUNCTIONS ==================== */
 
@@ -83,12 +85,25 @@ void bgpstream_elem_destroy(bgpstream_elem_t *elem) {
   bgpstream_community_set_destroy(elem->communities);
   elem->communities = NULL;
 
+#ifdef WITH_RTR
+  if(elem->annotations.active){
+    kh_destroy(rpki_result, elem->annotations.rpki_kh);
+    elem->annotations.rpki_kh = NULL;
+  }
+#endif
+
   free(elem);
 }
 
 void bgpstream_elem_clear(bgpstream_elem_t *elem) {
   bgpstream_as_path_clear(elem->aspath);
   bgpstream_community_set_clear(elem->communities);
+#ifdef WITH_RTR
+  if(elem->annotations.active && 
+     (elem->annotations.rpki_validation_status != BGPSTREAM_ELEM_RPKI_VALIDATION_STATUS_NOTVALIDATED)){
+    kh_clear(rpki_result, elem->annotations.rpki_kh);
+  }
+#endif
 }
 
 bgpstream_elem_t *bgpstream_elem_copy(bgpstream_elem_t *dst,
@@ -348,14 +363,16 @@ char *bgpstream_elem_custom_snprintf(char *buf, size_t len,
       if(B_FULL)
         return NULL;
 
-#if defined(FOUND_RTR)
+#ifdef WITH_RTR
       /* RPKI Validation */
-      char buf_rpki[1024];
-      c = bgpstream_elem_get_rpki_validation_result_snprintf(
-          buf_rpki, sizeof(buf_rpki), elem);
-      strcat(buf, buf_rpki);
-      written += c;
-      buf_p += c;
+      if(elem->annotations.active) {
+        char buf_rpki[BGPSTREAM_RPKI_RST_MAX_LEN];
+        c = bgpstream_elem_get_rpki_validation_result_snprintf(
+            buf_rpki, sizeof(buf_rpki), elem);
+        strcat(buf, buf_rpki);
+        written += c;
+        buf_p += c;
+      }
 #endif
       /* END OF LINE */
       break;
@@ -434,46 +451,26 @@ char *bgpstream_elem_snprintf(char *buf, size_t len,
   return bgpstream_elem_custom_snprintf(buf, len, elem, 1);
 }
 
-#if defined(FOUND_RTR)
+#ifdef WITH_RTR
 int bgpstream_elem_get_rpki_validation_result_snprintf(
     char *buf, size_t len, bgpstream_elem_t const *elem)
 {
-  char result_output[1024] = "";
+  int key;
+  char *val;
+  char result_output[BGPSTREAM_RPKI_RST_MAX_LEN];
+  char valid_prefixes[BGPSTREAM_RPKI_RST_MAX_LEN];
   if (elem->annotations.rpki_validation_status !=
       BGPSTREAM_ELEM_RPKI_VALIDATION_STATUS_NOTFOUND) {
     snprintf(result_output, sizeof(result_output), "%s%s", result_output,
              elem->annotations.rpki_validation_status ==
                      BGPSTREAM_ELEM_RPKI_VALIDATION_STATUS_INVALID
-                 ? "invalid;"
-                 : "valid;");
-    for (int i = 0; i < elem->annotations.rpki_validation_result.asn_used;
-         i++) {
-      char asn[1024];
-      snprintf(asn, sizeof(asn), "%" PRIu32 ",",
-               elem->annotations.rpki_validation_result.asn_pfx[i].asn);
-      strcat(result_output, asn);
-      for (int j = 0;
-           j < elem->annotations.rpki_validation_result.asn_pfx[i].pfx_used;
-           j++) {
-        char valid_prefix[INET6_ADDRSTRLEN];
-        bgpstream_pfx_snprintf(valid_prefix, INET6_ADDRSTRLEN,
-            (bgpstream_pfx_t *)&elem->annotations.rpki_validation_result
-                .asn_pfx[i].pfxs[j].pfx);
-        strcat(result_output, valid_prefix);
-        snprintf(asn, sizeof(asn), "-%" PRIu8,
-                 elem->annotations.rpki_validation_result.asn_pfx[i]
-                     .pfxs[j].max_pfx_len);
-        strcat(result_output, asn);
-        strcat(result_output,
-               j != elem->annotations.rpki_validation_result.asn_pfx[i]
-                               .pfx_used - 1
-                   ? " "
-                   : "");
-      }
-      strcat(result_output,
-             i != elem->annotations.rpki_validation_result.asn_used - 1 ? ";"
-                                                                        : "");
-    }
+                 ? "invalid;" : "valid;");
+
+    kh_foreach(elem->annotations.rpki_kh, key, val,
+      snprintf(valid_prefixes, sizeof(valid_prefixes), "%i,%s;", key, val);
+      strcat(result_output, valid_prefixes);
+    );
+    result_output[strlen(result_output) - 1] = 0;
   } else {
     snprintf(result_output, sizeof(result_output), "%s%s", result_output,
              "notfound");
@@ -482,17 +479,16 @@ int bgpstream_elem_get_rpki_validation_result_snprintf(
   return snprintf(buf, len, "%s", result_output);
 }
 
-void bgpstream_elem_get_rpki_validation_result(bgpstream_elem_t *elem,
+void bgpstream_elem_get_rpki_validation_result(struct rtr_mgr_config *cfg, bgpstream_elem_t *elem,
                                                char *prefix,
                                                uint32_t origin_asn,
                                                uint8_t mask_len)
 {
   if (elem->annotations.rpki_validation_status ==
       BGPSTREAM_ELEM_RPKI_VALIDATION_STATUS_NOTVALIDATED) {
-    cfg_tr = bgpstream_get_rtr_config();
 
     struct reasoned_result res_reasoned =
-        bgpstream_rtr_validate_reason(cfg_tr, origin_asn, prefix, mask_len);
+        bgpstream_rtr_validate_reason(cfg, origin_asn, prefix, mask_len);
 
     if (res_reasoned.result == BGP_PFXV_STATE_VALID) {
       elem->annotations.rpki_validation_status =
@@ -509,27 +505,43 @@ void bgpstream_elem_get_rpki_validation_result(bgpstream_elem_t *elem,
 
     if (elem->annotations.rpki_validation_status !=
         BGPSTREAM_ELEM_RPKI_VALIDATION_STATUS_NOTFOUND) {
-      bgpstream_rpki_validation_result_init(
-          &elem->annotations.rpki_validation_result, 2);
-      char valid_prefix[INET6_ADDRSTRLEN];
+
       char reason_prefix[INET6_ADDRSTRLEN];
+      char buf_p[BGPSTREAM_RPKI_RST_MAX_LEN];
+
+      int ret;
+      khiter_t k;
 
+      if(elem->annotations.khash_init != 1) {
+        elem->annotations.rpki_kh = kh_init(rpki_result);
+        elem->annotations.khash_init = 1; 
+      }
+  
       for (int i = 0; i < res_reasoned.reason_len; i++) {
-        bgpstream_rpki_validation_result_insert_asn(
-            &elem->annotations.rpki_validation_result,
-            res_reasoned.reason[i].asn);
-        lrtr_ip_addr_to_str(&(res_reasoned.reason[i].prefix), reason_prefix,
-                            sizeof(reason_prefix));
-        snprintf(valid_prefix, sizeof(valid_prefix), "%s/%" PRIu8, reason_prefix,
-                 res_reasoned.reason[i].min_len);
-
-        bgpstream_pfx_t pfx;
-        bgpstream_str2pfx(valid_prefix, (bgpstream_pfx_storage_t *)&pfx);
-        bgpstream_rpki_validation_result_insert_pfx(
-            &elem->annotations.rpki_validation_result,
-            res_reasoned.reason[i].asn, &pfx, res_reasoned.reason[i].max_len);
+	      if(kh_get(rpki_result, elem->annotations.rpki_kh, res_reasoned.reason[i].asn) == 
+           kh_end(elem->annotations.rpki_kh)){
+       	  k = kh_put(rpki_result, elem->annotations.rpki_kh, (int) res_reasoned.reason[i].asn, &ret);
+	        kh_val(elem->annotations.rpki_kh, k) = '\0';
+        }  
+        else {
+          k = kh_get(rpki_result, elem->annotations.rpki_kh, (int) res_reasoned.reason[i].asn);
+        }
+
+        lrtr_ip_addr_to_str(&(res_reasoned.reason[i].prefix), reason_prefix, sizeof(reason_prefix));
+        snprintf(elem->annotations.valid_prefix[k], BGPSTREAM_RPKI_RST_MAX_LEN, "%s/%" PRIu8 "-%"PRIu8, 
+                 reason_prefix, res_reasoned.reason[i].min_len, res_reasoned.reason[i].max_len);
+
+        if(!kh_val(elem->annotations.rpki_kh, k)){
+          kh_val(elem->annotations.rpki_kh, k) = elem->annotations.valid_prefix[k];
+        }
+        else if(!strstr(kh_val(elem->annotations.rpki_kh, k), elem->annotations.valid_prefix[k])) {
+          snprintf(buf_p, sizeof(buf_p), "%s %s", kh_val(elem->annotations.rpki_kh, k), 
+                   elem->annotations.valid_prefix[k]);
+          kh_val(elem->annotations.rpki_kh, k) = buf_p;
+        }
       }
     }
+
     free(res_reasoned.reason);
   }
 }