RPKI: Importing routes into roa table

Implementation based on RTRLib.

Communication between rtrlib threads and main thread through pipe()
  sockets and notify list like in BFD protocol.

TODO:
  - load rtrlib using dlopen()
  - take into account preferences of cache servers in configuration

conf/confbase.Y

@@ -26,6 +26,13 @@ CF_HDR
 
 CF_DEFINES
 
+static void
+check_u8(unsigned val)
+{
+  if (val > 0xFF)
+    cf_error("Value %d out of range (0-255)", val);
+}
+
 static void
 check_u16(unsigned val)
 {
@@ -55,6 +62,7 @@ CF_DECLS
   struct lsadb_show_data *ld;
   struct iface *iface;
   struct roa_table *rot;
+  struct roa_table_config *rotcf;
   void *g;
   bird_clock_t time;
   struct prefix px;

nest/config.Y

@@ -76,6 +76,7 @@ CF_ENUM(T_ENUM_ROA, ROA_, UNKNOWN, VALID, INVALID)
 %type <ra> r_args
 %type <ro> roa_args
 %type <rot> roa_table_arg
+%type <rotcf> roa_table_cf
 %type <sd> sym_args
 %type <i> proto_start echo_mask echo_size debug_mask debug_list debug_flag mrtdump_mask mrtdump_list mrtdump_flag export_mode roa_mode limit_action tab_sorted tos
 %type <ps> proto_patt proto_patt2
@@ -262,6 +263,14 @@ rtable:
    }
  ;
 
+roa_table_cf:
+    SYM {
+      if ($1->class != SYM_ROA) cf_error("ROA table name expected");
+      $$ = $1->def;
+    }
+  ;
+
+
 CF_ADDTO(conf, debug_default)
 
 debug_default:

nest/proto.c

@@ -921,6 +921,7 @@ protos_build(void)
 #endif
 #ifdef CONFIG_RPKI
   proto_build(&proto_rpki);
+  rpki_init_all();
 #endif
 
   proto_pool = rp_new(&root_pool, "Protocols");

nest/route.h

@@ -600,6 +600,7 @@ struct roa_show_data {
 #define ROA_SRC_ANY	0
 #define ROA_SRC_CONFIG	1
 #define ROA_SRC_DYNAMIC	2
+#define ROA_SRC_RPKI    3
 
 #define ROA_SHOW_ALL	0
 #define ROA_SHOW_PX	1

proto/rpki/config.Y

@@ -25,7 +25,13 @@ CF_GRAMMAR
 CF_ADDTO(proto, rpki_proto)
 
 rpki_proto:
-   rpki_proto_start proto_name '{' rpki_proto_opts '}'
+   rpki_proto_start proto_name '{' rpki_proto_opts '}' {
+     if (RPKI_CFG->roa_table_cf == NULL)
+     {
+       cf_error("For the RPKI protocol must be specified a roa table");
+     }
+   }
+ ;
 
 rpki_proto_start:
    proto_start RPKI {
@@ -42,6 +48,7 @@ rpki_proto_opts:
 rpki_proto_item:
    proto_item
  | CACHE LIST '{' rpki_cache_list '}'
+ | ROA TABLE roa_table_cf { RPKI_CFG->roa_table_cf = $3; }
  ;
 
 rpki_cache_list:
@@ -70,8 +77,8 @@ rpki_cache_opts:
  ;
 
 rpki_cache_opts_item:
-   PORT expr 		{ bsnprintf(this_rpki_cache->port, RPKI_PORT_MAX_LENGTH_STR, "%u", (u16) $2); }
+   PORT expr 		{ check_u16($2); bsnprintf(this_rpki_cache->port, RPKI_PORT_MAX_LENGTH_STR, "%u", (u16) $2); }
- | PREFERENCE expr	{ this_rpki_cache->preference = $2; }
+ | PREFERENCE expr	{ check_u8($2); this_rpki_cache->preference = $2; }
  ;
 
 CF_CODE

proto/rpki/rpki.c

@@ -14,67 +14,158 @@
 
 #include <stdlib.h>
 #include <unistd.h>
+#include <pthread.h>
 
 #include "proto/rpki/rpki.h"
 #include "lib/socket.h"
+#include "lib/ip.h"
+#include "nest/route.h"
 
-struct proto *ugly_hack_to_get_proto;
+struct rpki_entry {
+  node n;
+  u32 asn;
+  ip_addr ip;
+  u8 min_len;
+  u8 max_len;
+  u8 added;
+  struct rpki_proto *rpki;
+};
 
-static void status_cb(const struct rtr_mgr_group *group, enum rtr_mgr_status status, const struct rtr_socket *socket, void *data)
+void pipe_drain(int fd); 	/* implementation in io.c */
+void pipe_kick(int fd); 	/* implementation in io.c */
+
+static list rpki_proto_list;
+
+void
+rpki_init_all(void)
 {
-  struct rpki_proto *rpki = data;
+  init_list(&rpki_proto_list);
-  if(status == RTR_MGR_ERROR)
-  {
-    RPKI_TRACE(rpki, "Error -> Should we here stop the protocol?"); /* FIXME */
-  }
-
-  RPKI_TRACE(rpki, "Status: %s\t%s", rtr_mgr_status_to_str(status), rtr_state_to_str(socket->state));
 }
 
-static void update_cb(struct pfx_table *p, const struct pfx_record rec, const bool added)
+static void
+status_cb(const struct rtr_mgr_group *group, enum rtr_mgr_status status, const struct rtr_socket *socket, void *data)
 {
-  /* FIXME: update_cb() should have void *data attribute, same like status_cb() */
+  struct rpki_proto *p = data;
-  struct proto *P = ugly_hack_to_get_proto;
-  struct rpki_proto *rpki = (struct rpki_proto *) P;
 
-  ip4_addr ip4 = {};
+  if (status == RTR_MGR_ERROR)
-  ip6_addr ip6 = {};
-  char ip[INET6_ADDRSTRLEN];
-  if (rec.prefix.ver == RTRLIB_IPV4)
   {
-    ip4 = ipa_from_u32(rec.prefix.u.addr4.addr);
+    RPKI_ERROR(p, "Error -> Should we here stop the protocol?"); /* FIXME */
-    ip4_ntop(ip4, ip);
   }
   else
   {
-    ip6 = ip6_build(rec.prefix.u.addr6.addr[0], rec.prefix.u.addr6.addr[1], rec.prefix.u.addr6.addr[2], rec.prefix.u.addr6.addr[3]);
+    RPKI_TRACE(p, "status: %s\t%s", rtr_mgr_status_to_str(status), rtr_state_to_str(socket->state));
-    ip6_ntop(ip6, ip);
   }
+}
 
-  if(added)
+static void
+send_data_to_main_thread(struct rpki_proto *p, struct rpki_entry *e)
+{
+  rpki_lock_sessions(p);
+  add_tail(&p->notify_list, &e->n);
+  rpki_unlock_sessions(p);
+
+  pipe_kick(p->notify_write_sk->fd);
+}
+
+static void
+log_skip_entry(struct rpki_proto *p, const struct pfx_record *rec, const bool added)
+{
+  char ip_buf[INET6_ADDRSTRLEN];
+  ip4_addr ip4;
+  ip6_addr ip6;
+
+  if (rec->prefix.ver == RTRLIB_IPV4)
   {
-    RPKI_TRACE(rpki, "+++ %45s/%u-%-3u \tASN: %10u", ip, rec.min_len, rec.max_len, rec.asn);
+    ip4 = ipa_from_u32(rec->prefix.u.addr4.addr);
-//  P->rte_insert();
+    ip4_ntop(ip4, ip_buf);
   }
   else
   {
-    RPKI_TRACE(rpki, "--- %45s/%u-%-3u \tASN: %10u", ip, rec.min_len, rec.max_len, rec.asn);
+    ip6 = ip6_build(rec->prefix.u.addr6.addr[0], rec->prefix.u.addr6.addr[1], rec->prefix.u.addr6.addr[2], rec->prefix.u.addr6.addr[3]);
-//  P->rte_remove();
+    ip6_ntop(ip6, ip_buf);
   }
+
+#define LOG_SKIP_ENTRY_FMT(operation_name) "skip unsupported IP version: " operation_name " %25s/%u-%-3u \tASN: %10u"
+
+  if (added)
+  {
+    RPKI_TRACE(p, LOG_SKIP_ENTRY_FMT("add"), ip_buf, rec->min_len, rec->max_len, rec->asn);
+  }
+  else
+  {
+    RPKI_TRACE(p, LOG_SKIP_ENTRY_FMT("del"), ip_buf, rec->min_len, rec->max_len, rec->asn);
+  }
+}
+
+static struct rpki_proto *
+find_rpki_proto_by_rtr_socket(const struct rtr_socket *socket)
+{
+  struct rpki_proto *p_not_skipped_back;
+  unsigned int i, j;
+
+  WALK_LIST(p_not_skipped_back, rpki_proto_list)
+  {
+    struct rpki_proto *p = SKIP_BACK(struct rpki_proto, rpki_node, p_not_skipped_back);
+
+    for(i = 0; i < p->rtr_conf->len; i++)
+    {
+      for(j = 0; j < p->rtr_conf->groups[i].sockets_len; j++)
+      {
+	if (socket == p->rtr_conf->groups[i].sockets[j])
+	  return p;
+      }
+    }
+  }
+
+  return NULL;
+}
+
+static void
+rtr_thread_update_hook(struct pfx_table *pfx_table, const struct pfx_record rec, const bool added)
+{
+  struct rpki_proto *p = find_rpki_proto_by_rtr_socket(rec.socket);
+
+  /* process only records that are the same with BIRD IP version */
+#ifdef IPV6
+  if (rec.prefix.ver != RTRLIB_IPV6)
+  {
+    log_skip_entry(p, &rec);
+    return;
+  }
+#else
+  if (rec.prefix.ver != RTRLIB_IPV4)
+  {
+    log_skip_entry(p, &rec, added);
+    return;
+  }
+#endif
+
+#ifdef IPV6
+  ip_addr ip = ip6_build(rec.prefix.u.addr6.addr[0], rec.prefix.u.addr6.addr[1], rec.prefix.u.addr6.addr[2], rec.prefix.u.addr6.addr[3]);
+#else
+  ip_addr ip = ipa_from_u32(rec.prefix.u.addr4.addr);
+#endif
+
+  struct rpki_entry *e = mb_allocz(p->p.pool, sizeof(struct rpki_entry));
+  e->added = added;
+  e->asn = rec.asn;
+  e->ip = ip;
+  e->max_len = rec.max_len;
+  e->min_len = rec.min_len;
+  e->rpki = p;
+
+  send_data_to_main_thread(p, e);
 }
 
 static struct proto *
 rpki_init(struct proto_config *C)
 {
   struct proto *P = proto_new(C, sizeof(struct rpki_proto));
-  struct rpki_proto *rpki = (struct rpki_proto *) P;
+  struct rpki_proto *p = (struct rpki_proto *) P;
-  struct rpki_config *cf = (struct rpki_config *) C;
+  p->cf = (struct rpki_config *) C;
 
-  RPKI_TRACE(rpki, "------------- rpki_init -------------");
+  RPKI_TRACE(p, "------------- rpki_init -------------");
 
-  ugly_hack_to_get_proto = P;
-
-  /* TODO: Add defaults */
   return P;
 }
 
@@ -83,6 +174,7 @@ rpki_new_cache(void)
 {
   struct rpki_cache *cache = (struct rpki_cache *)cfg_allocz(sizeof(struct rpki_cache));
   strcpy(cache->port, RPKI_PORT);
+  cache->preference = ~0;
   return cache;
 }
 
@@ -100,22 +192,102 @@ normalize_fulfillment_of_cache(struct rpki_cache *cache)
   bzero(&cache->tr_tcp, sizeof(struct tr_socket));
 }
 
+static int
+rpki_notify_hook(struct birdsock *sk, int size)
+{
+  struct rpki_proto *p = sk->data;
+  struct rpki_entry *entry;
+
+  pipe_drain(sk->fd);
+
+  rpki_lock_sessions(p);
+  /* TODO: optimize like in the BFD proto */
+  WALK_LIST_FIRST(entry, p->notify_list)
+  {
+    rem2_node(&entry->n);
+    if (entry->added)
+      roa_add_item(p->cf->roa_table_cf->table, entry->ip, entry->min_len, entry->max_len, entry->asn, ROA_SRC_RPKI);
+    else
+      roa_delete_item(p->cf->roa_table_cf->table, entry->ip, entry->min_len, entry->max_len, entry->asn, ROA_SRC_RPKI);
+  }
+  rpki_unlock_sessions(p);
+}
+
+static void
+rpki_noterr_hook(struct birdsock *sk, int err)
+{
+  struct rpki_proto *p = sk->data;
+  RPKI_ERROR(p, "Notify socket error: %m", err);
+}
+
+static void
+create_read_socket(struct rpki_proto *p, int fd)
+{
+  sock *sk = sk_new(p->p.pool);
+  sk->type = SK_MAGIC;
+  sk->fd = fd;
+  sk->rx_hook = rpki_notify_hook;
+  sk->err_hook = rpki_noterr_hook;
+  sk->data = p;
+  if (sk_open(sk) < 0)
+    RPKI_DIE(p, "read socket sk_open() failed");
+  p->notify_read_sk = sk;
+}
+
+static void
+create_write_socket(struct rpki_proto *p, int fd)
+{
+  sock *sk = sk_new(p->p.pool);
+  sk->type = SK_MAGIC;
+  sk->fd = fd;
+  sk->flags = SKF_THREAD;
+  sk->data = p;
+  if (sk_open(sk) < 0)
+    RPKI_DIE(p, "write socket sk_open() failed");
+  p->notify_write_sk = sk;
+}
+
+static void
+create_rw_sockets(struct rpki_proto *p)
+{
+  int pipe_fildes[2];
+
+  int rv = pipe(pipe_fildes);
+  if (rv < 0)
+    RPKI_DIE(p, "pipe: %m");
+
+  create_read_socket (p, pipe_fildes[0]);
+  create_write_socket(p, pipe_fildes[1]);
+}
+
 static int
 rpki_start(struct proto *P)
 {
-  struct rpki_proto *rpki = (struct rpki_proto *) P;
+  struct rpki_proto *p = (struct rpki_proto *) P;
   struct rpki_config *cf = (struct rpki_config *) (P->cf);
 
-  RPKI_TRACE(rpki, "------------- rpki_start -------------");
+  RPKI_TRACE(p, "------------- rpki_start -------------");
 
-  rpki->rtr_groups_len = get_list_length(&cf->cache_list);
+  create_rw_sockets(p);
-  rpki->rtr_groups = mb_allocz(P->pool, rpki->rtr_groups_len * sizeof(struct rtr_mgr_group));
+  init_list(&p->notify_list);
-  struct rtr_mgr_group *groups = rpki->rtr_groups;
+  pthread_spin_init(&p->notify_lock, PTHREAD_PROCESS_PRIVATE);
+
+  add_tail(&rpki_proto_list, &p->rpki_node);
+
+  p->rtr_sockets_len = get_list_length(&cf->cache_list);
+  p->rtr_groups = mb_allocz(P->pool, 1 * sizeof(struct rtr_mgr_group));
+  struct rtr_mgr_group *groups = p->rtr_groups;
+
+  p->rtr_sockets = mb_allocz(P->pool, p->rtr_sockets_len * sizeof(struct rtr_socket *));
+  groups[0].sockets = p->rtr_sockets;
+  groups[0].sockets_len = p->rtr_sockets_len;
+  groups[0].preference = 1;
 
   uint idx = 0;
   struct rpki_cache *cache;
   WALK_LIST(cache, cf->cache_list)
   {
+    /* TODO: Make them dynamic. Reconfigure reallocate structures */
     struct tr_tcp_config *tcp_config = &cache->tcp_config;
     struct rtr_socket *rtr_tcp = &cache->rtr_tcp;
     struct tr_socket *tr_tcp = &cache->tr_tcp;
@@ -129,16 +301,13 @@ rpki_start(struct proto *P)
     // create an rtr_socket and associate it with the transport socket
     rtr_tcp->tr_socket = tr_tcp;
 
-    groups[idx].sockets = mb_allocz(P->pool, 1 * sizeof(struct rtr_socket *));
+    groups[0].sockets[idx] = rtr_tcp;
-    groups[idx].sockets_len = 1;
-    groups[idx].sockets[0] = rtr_tcp;
-    groups[idx].preference = cache->preference;
 
     idx++;
   }
 
-  rpki->rtr_conf = rtr_mgr_init(groups, rpki->rtr_groups_len, 30, 520, &update_cb, NULL, &status_cb, rpki);
+  p->rtr_conf = rtr_mgr_init(groups, 1, 30, 520, &rtr_thread_update_hook, NULL, &status_cb, p);
-  rtr_mgr_start(rpki->rtr_conf);
+  rtr_mgr_start(p->rtr_conf);
 
   return PS_UP;
 }
@@ -146,15 +315,12 @@ rpki_start(struct proto *P)
 static int
 rpki_shutdown(struct proto *P)
 {
-  struct rpki_proto *rpki = (struct rpki_proto *) P;
+  struct rpki_proto *p = (struct rpki_proto *) P;
-  struct rpki_config *cf = (struct rpki_config *) (P->cf);
 
-  RPKI_TRACE(rpki, "------------- rpki_shutdown -------------");
+  rtr_mgr_stop(p->rtr_conf);
-
+  rtr_mgr_free(p->rtr_conf);
-  rtr_mgr_stop(rpki->rtr_conf);
+  mb_free(p->rtr_groups);
-  rtr_mgr_free(rpki->rtr_conf);
+  mb_free(p->rtr_sockets);
-
-  /* TODO: fix memory leaks created by start->disable->enable rpki protocol */
 
   return PS_DOWN;
 }

proto/rpki/rpki.h

@@ -9,25 +9,42 @@
 #ifndef _BIRD_RPKI_H_
 #define _BIRD_RPKI_H_
 
+#include <pthread.h>
+
 #include "rtrlib/rtrlib.h"
 
 #include "nest/bird.h"
 #include "nest/protocol.h"
+#include "lib/socket.h"
 
 #define RPKI_PORT "8282"
 #define RPKI_PORT_MAX_LENGTH_STR 6
+#define RPKI_RX_BUFFER_EXT_SIZE 0xffff
+#define RPKI_TX_BUFFER_EXT_SIZE 0xffff
 
-#define RPKI_TRACE(rpki, msg, args...) 					\
+#define RPKI_LOG(log_level, p, msg, args...) 				\
   do { 									\
-    if (rpki->p.debug)							\
+    log(log_level "%s: " msg, p->p.name , ## args); 			\
-      log(L_TRACE "%s: " msg, rpki->p.name , ## args ); 		\
   } while(0)
 
+#define RPKI_TRACE(p, msg, args...) 					\
+  do {									\
+    if (p->p.debug)							\
+      RPKI_LOG(L_TRACE, p, msg, ## args);				\
+  } while(0)
+
+#define RPKI_ERROR(p, msg, args...) RPKI_LOG(L_ERR, p, msg, ## args);
+#define RPKI_DIE(p, msg, args...) 					\
+    do {								\
+      RPKI_LOG(L_FATAL, p, msg, ## args);				\
+      exit(1);								\
+    } while(0)
+
 struct rpki_cache {
   node n;		/* in struct rpki_config.cache_list */
   ip_addr ip;
   char *full_domain_name;
-  char port[RPKI_PORT_MAX_LENGTH_STR]; /* the highest port is "65535" */
+  char port[RPKI_PORT_MAX_LENGTH_STR]; /* TODO change to u16 */
   u8 preference;
 
   /* below are private variables */
@@ -36,20 +53,35 @@ struct rpki_cache {
   struct tr_socket tr_tcp;
   struct tr_tcp_config tcp_config;
   char ip_buf[INET6_ADDRSTRLEN];
+  char port_buf[RPKI_PORT_MAX_LENGTH_STR]; /* the highest port is "65535" */
 };
 
 struct rpki_config {
   struct proto_config c;
   list cache_list; 	/* (struct rpki_cache *) */
+  struct roa_table_config *roa_table_cf;
 };
 
 struct rpki_proto {
   struct proto p;
+  node rpki_node;
+  struct rpki_config *cf;
   struct rtr_mgr_config *rtr_conf;
   struct rtr_mgr_group *rtr_groups;
-  uint rtr_groups_len;
+  struct rtr_socket **rtr_sockets;
+  uint rtr_sockets_len;
+
+  sock *notify_read_sk;
+  sock *notify_write_sk;
+  list notify_list;
+  pthread_spinlock_t notify_lock;
 };
 
 struct rpki_cache *rpki_new_cache(void);
 
+static inline void rpki_lock_sessions(struct rpki_proto *p) { pthread_spin_lock(&p->notify_lock); }
+static inline void rpki_unlock_sessions(struct rpki_proto *p) { pthread_spin_unlock(&p->notify_lock); }
+
+void rpki_init_all(void);
+
 #endif /* _BIRD_RPKI_H_ */

rtrlib

@@ -1 +1 @@
-Subproject commit 6fcc554a75494cee701e13db2791a7fad2eb8813
+Subproject commit ee04ff8280baba1fc5e6e41ba7bd66bf3142d81e