Add support for End-of-RIB marker elems

.gitignore

@@ -27,6 +27,7 @@ tools/bgpreader
 test/bgpstream-test
 test/bgpstream-test-filters
 test/bgpstream-test-rislive
+test/bgpstream-test-eor
 test/bgpstream-test-rpki
 test/bgpstream-test-utils-addr
 test/bgpstream-test-utils-pfx

configure.ac

@@ -29,7 +29,7 @@ AC_PREREQ([2.68])
 
 # bgpstream package version
 m4_define([PKG_MAJOR_VERSION], [2])
-m4_define([PKG_MID_VERSION],   [3])
+m4_define([PKG_MID_VERSION],   [4])
 m4_define([PKG_MINOR_VERSION], [0])
 
 AC_INIT([libbgpstream], PKG_MAJOR_VERSION.PKG_MID_VERSION.PKG_MINOR_VERSION, [bgpstream-info@caida.org])

lib/bgpstream_bgpdump.c

@@ -67,6 +67,16 @@ char *bgpstream_record_elem_bgpdump_snprintf(char *buf, size_t len,
   ssize_t c = 0;      /* < how many chars were written */
   char *buf_p = buf;
 
+  /* The original bgpdump tool does not represent End-of-RIB markers in its
+   * output (an EoR is an empty UPDATE, which it silently drops). To stay
+   * faithful to that format we emit nothing for them. Callers should skip empty
+   * output. */
+  if (elem->type == BGPSTREAM_ELEM_TYPE_END_OF_RIB) {
+    if (len > 0)
+      buf[0] = '\0';
+    return buf;
+  }
+
   /* Record type */
   switch (elem->type) {
   case BGPSTREAM_ELEM_TYPE_RIB:

lib/bgpstream_elem.c

@@ -92,6 +92,15 @@ void bgpstream_elem_clear(bgpstream_elem_t *elem)
 {
   bgpstream_as_path_clear(elem->as_path);
   bgpstream_community_set_clear(elem->communities);
+
+  /* reset the remaining type-dependent fields so that a re-used elem does not
+   * leak stale values (e.g. path attributes from a previous elem) */
+  elem->nexthop.version = BGPSTREAM_ADDR_VERSION_UNKNOWN;
+  elem->has_origin = 0;
+  elem->has_med = 0;
+  elem->has_local_pref = 0;
+  elem->atomic_aggregate = 0;
+  elem->aggregator.has_aggregator = 0;
 }
 
 bgpstream_elem_t *bgpstream_elem_copy(bgpstream_elem_t *dst,
@@ -129,6 +138,7 @@ int bgpstream_elem_type_snprintf(char *buf, size_t len,
     case BGPSTREAM_ELEM_TYPE_ANNOUNCEMENT: ch = 'A'; break;
     case BGPSTREAM_ELEM_TYPE_WITHDRAWAL:   ch = 'W'; break;
     case BGPSTREAM_ELEM_TYPE_PEERSTATE:    ch = 'S'; break;
+    case BGPSTREAM_ELEM_TYPE_END_OF_RIB:   ch = 'E'; break;
     default:
       if (len > 0)
         buf[0] = '\0';
@@ -378,6 +388,30 @@ char *bgpstream_elem_custom_snprintf(char *buf, size_t len,
     /* END OF LINE */
     break;
 
+  case BGPSTREAM_ELEM_TYPE_END_OF_RIB:
+
+    /* PREFIX (AFI indicator) */
+    if (bgpstream_pfx_snprintf(buf_p, B_REMAIN, &(elem->prefix)) == NULL) {
+      if (errno != ENOSPC)
+        bgpstream_log(BGPSTREAM_LOG_ERR, "Malformed prefix (E)");
+      return NULL;
+    }
+    SEEK_STR_END;
+    ADD_PIPE;
+    /* NEXT HOP (empty) */
+    ADD_PIPE;
+    /* AS PATH (empty) */
+    ADD_PIPE;
+    /* ORIGIN AS (empty) */
+    ADD_PIPE;
+    /* COMMUNITIES (empty) */
+    ADD_PIPE;
+    /* OLD STATE (empty) */
+    ADD_PIPE;
+    /* NEW STATE (empty) */
+    /* END OF LINE */
+    break;
+
   default:
     bgpstream_log(BGPSTREAM_LOG_ERR, "Error during elem processing");
     return NULL;

lib/bgpstream_elem.h

@@ -124,6 +124,9 @@ typedef enum {
   /** Peer state change */
   BGPSTREAM_ELEM_TYPE_PEERSTATE = 4,
 
+  /** End-of-RIB marker (RFC 4724) */
+  BGPSTREAM_ELEM_TYPE_END_OF_RIB = 5,
+
 } bgpstream_elem_type_t;
 
 typedef struct struct_bgpstream_annotations_t {

lib/bgpstream_filter.c

@@ -123,6 +123,8 @@ int bgpstream_filter_mgr_filter_add(bgpstream_filter_mgr_t *this,
       this->elemtype_mask |= (BGPSTREAM_FILTER_ELEM_TYPE_WITHDRAWAL);
     } else if (strcmp(filter_value, "peerstates") == 0) {
       this->elemtype_mask |= (BGPSTREAM_FILTER_ELEM_TYPE_PEERSTATE);
+    } else if (strcmp(filter_value, "endofrib") == 0) {
+      this->elemtype_mask |= (BGPSTREAM_FILTER_ELEM_TYPE_END_OF_RIB);
     } else {
       bgpstream_log(BGPSTREAM_LOG_ERR,
                     "unknown element type '%s'", filter_value);

lib/bgpstream_filter.h

@@ -41,6 +41,7 @@
 #define BGPSTREAM_FILTER_ELEM_TYPE_ANNOUNCEMENT 0x2
 #define BGPSTREAM_FILTER_ELEM_TYPE_WITHDRAWAL 0x4
 #define BGPSTREAM_FILTER_ELEM_TYPE_PEERSTATE 0x8
+#define BGPSTREAM_FILTER_ELEM_TYPE_END_OF_RIB 0x10
 
 /* hash table community filter:
  * community -> filter mask (asn only, value only, both) */

lib/bgpstream_record.c

@@ -155,6 +155,11 @@ static int elem_check_filters(bgpstream_record_t *record,
         !(filter_mgr->elemtype_mask & BGPSTREAM_FILTER_ELEM_TYPE_WITHDRAWAL)) {
       return 0;
     }
+
+    if (elem->type == BGPSTREAM_ELEM_TYPE_END_OF_RIB &&
+        !(filter_mgr->elemtype_mask & BGPSTREAM_FILTER_ELEM_TYPE_END_OF_RIB)) {
+      return 0;
+    }
   }
 
   /* Checking peer ASNs: if the filter is on and the peer asn is not in the

lib/formats/bgpstream_parsebgp_common.c

@@ -310,6 +310,65 @@ static int handle_prefix(bgpstream_elem_t *elem,
     }                                                                          \
   } while (0)
 
+// Detect an RFC 4724 End-of-RIB marker and, if found, populate `elem` as an
+// END_OF_RIB elem whose prefix encodes the AFI (0.0.0.0/0 for IPv4 unicast,
+// ::/0 for IPv6 unicast). Returns 1 if an EoR was detected and `elem` was
+// populated, 0 otherwise.
+static int handle_end_of_rib(bgpstream_elem_t *elem,
+                             parsebgp_bgp_update_t *update)
+{
+  // a 16-byte zeroed source so we can build the 0.0.0.0/0 or ::/0 prefix that
+  // encodes the EoR AFI
+  static const uint8_t zero_addr[16] = { 0 };
+
+  // IPv4 unicast EoR: a completely empty UPDATE (no withdrawn or announced
+  // NLRIs, and no path attributes). Note that we must check the original NLRI
+  // counts (not the upd_state counters, which have been decremented to zero as
+  // elems were yielded) so that we don't mistake a withdrawal-only UPDATE for
+  // an EoR.
+  if (update->path_attrs.attrs_cnt == 0 &&
+      update->withdrawn_nlris.prefixes_cnt == 0 &&
+      update->announced_nlris.prefixes_cnt == 0) {
+    bgpstream_elem_clear(elem);
+    elem->type = BGPSTREAM_ELEM_TYPE_END_OF_RIB;
+    bgpstream_ipv4_addr_init(&elem->prefix.address, zero_addr);
+    elem->prefix.mask_len = 0;
+    return 1;
+  }
+
+  // IPv4/IPv6 unicast EoR: the UPDATE carries exactly one path attribute, an
+  // empty MP_UNREACH_NLRI (zero withdrawn NLRIs) for the IPv4 or IPv6 unicast
+  // address family. Per RFC 4724, a valid EoR must not carry any native NLRIs
+  // either, so reject UPDATEs that contained IPv4 withdrawals or announcements.
+  if (update->path_attrs.attrs_cnt != 1 ||
+      update->withdrawn_nlris.prefixes_cnt != 0 ||
+      update->announced_nlris.prefixes_cnt != 0) {
+    return 0;
+  }
+  parsebgp_bgp_update_path_attr_t *mp_unreach_attr =
+    &update->path_attrs.attrs[PARSEBGP_BGP_PATH_ATTR_TYPE_MP_UNREACH_NLRI];
+  if (mp_unreach_attr->type != PARSEBGP_BGP_PATH_ATTR_TYPE_MP_UNREACH_NLRI) {
+    return 0;
+  }
+  parsebgp_bgp_update_mp_unreach_t *mp_unreach = mp_unreach_attr->data.mp_unreach;
+  if (mp_unreach->withdrawn_nlris_cnt != 0 ||
+      mp_unreach->safi != PARSEBGP_BGP_SAFI_UNICAST) {
+    return 0;
+  }
+  bgpstream_elem_clear(elem);
+  elem->type = BGPSTREAM_ELEM_TYPE_END_OF_RIB;
+  if (mp_unreach->afi == PARSEBGP_BGP_AFI_IPV6) {
+    bgpstream_ipv6_addr_init(&elem->prefix.address, zero_addr);
+  } else if (mp_unreach->afi == PARSEBGP_BGP_AFI_IPV4) {
+    bgpstream_ipv4_addr_init(&elem->prefix.address, zero_addr);
+  } else {
+    // some other (unsupported) address family - not an EoR we represent
+    return 0;
+  }
+  elem->prefix.mask_len = 0;
+  return 1;
+}
+
 int bgpstream_parsebgp_process_update(bgpstream_parsebgp_upd_state_t *upd_state,
                                       bgpstream_elem_t *elem,
                                       parsebgp_bgp_msg_t *bgp)
@@ -325,6 +384,11 @@ int bgpstream_parsebgp_process_update(bgpstream_parsebgp_upd_state_t *upd_state,
       return 0;
     }
 
+    // the message claims to be an UPDATE, but carries no UPDATE payload
+    if (update == NULL) {
+      return 0;
+    }
+
     // how many native withdrawals will we process?
     upd_state->withdrawal_v4_cnt = update->withdrawn_nlris.prefixes_cnt;
 
@@ -355,6 +419,16 @@ int bgpstream_parsebgp_process_update(bgpstream_parsebgp_upd_state_t *upd_state,
   // are we at end-of-elems?
   if (upd_state->withdrawal_v4_cnt == 0 && upd_state->withdrawal_v6_cnt == 0 &&
       upd_state->announce_v4_cnt == 0 && upd_state->announce_v6_cnt == 0) {
+
+    // Check if this is an End-of-RIB marker (RFC 4724). We only emit a single
+    // EoR elem per record, so guard with eor_done.
+    if (upd_state->eor_done == 0 && update != NULL) {
+      upd_state->eor_done = 1;
+      if (handle_end_of_rib(elem, update) != 0) {
+        return 1;
+      }
+    }
+
     return 0;
   }
 

lib/formats/bgpstream_parsebgp_common.h

@@ -86,6 +86,9 @@ typedef struct bgpstream_parsebgp_upd_state {
   // has the BGP4MP state been prepared
   int ready;
 
+  // has the End-of-RIB check been performed
+  int eor_done;
+
   // how many native (IPv4) withdrawals still to yield
   int withdrawal_v4_cnt;
   int withdrawal_v4_idx;

lib/formats/bs_format_rislive.c

@@ -682,6 +682,7 @@ void bs_format_rislive_clear_data(bgpstream_format_t *format, void *data)
   bgpstream_elem_clear(rd->elem);
   rd->end_of_elems = 0;
   rd->next_re = 0;
+  rd->msg_type = 0;
   bgpstream_parsebgp_upd_state_reset(&rd->upd_state);
   parsebgp_clear_msg(rd->msg);
 }

test/Makefile.am

@@ -36,6 +36,7 @@ TESTS = 				\
 	bgpstream-test			\
 	bgpstream-test-filters		\
 	bgpstream-test-rislive		\
+	bgpstream-test-eor		\
 	bgpstream-test-utils-addr	\
 	bgpstream-test-utils-pfx	\
 	bgpstream-test-utils-patricia	\
@@ -46,6 +47,7 @@ check_PROGRAMS = 			\
 	bgpstream-test			\
 	bgpstream-test-filters		\
 	bgpstream-test-rislive		\
+	bgpstream-test-eor		\
 	bgpstream-test-utils-addr	\
 	bgpstream-test-utils-pfx	\
 	bgpstream-test-utils-patricia	\
@@ -56,6 +58,7 @@ check_PROGRAMS = 			\
 EXTRA_DIST = 	sqlite_test.db \
 		csv_test.csv \
 		ris-live-stream.json \
+		eor-stream.json \
 		routeviews.route-views.jinx.ribs.1427846400.bz2 \
 		routeviews.route-views.jinx.updates.1427846400.bz2 \
 		ris.rrc06.updates.1427846400.gz \
@@ -70,6 +73,9 @@ bgpstream_test_filters_LDADD   = $(top_builddir)/lib/libbgpstream.la
 bgpstream_test_rislive_SOURCES = bgpstream-test-rislive.c bgpstream_test.h
 bgpstream_test_rislive_LDADD   = $(top_builddir)/lib/libbgpstream.la
 
+bgpstream_test_eor_SOURCES = bgpstream-test-eor.c bgpstream_test.h
+bgpstream_test_eor_LDADD   = $(top_builddir)/lib/libbgpstream.la
+
 bgpstream_test_rpki_SOURCES = bgpstream-test-rpki.c bgpstream-test-rpki.h bgpstream_test.h
 bgpstream_test_rpki_LDADD   = $(top_builddir)/lib/libbgpstream.la