From ddf7cc5a4de8ed1bb9d5fcab7874716392a96415 Mon Sep 17 00:00:00 2001 From: Christian Giese Date: Wed, 16 Mar 2022 14:39:30 +0100 Subject: [PATCH] BGP enhancements + add tcp idle callback function + extend bgp-sessions command + log update start/stop timestamp + decode raw update stream + ... --- code/bngblaster/src/bbl_tcp.c | 7 ++- code/bngblaster/src/bbl_tcp.h | 5 +- code/bngblaster/src/bgp/bgp.c | 72 +++++++++++++++++++++++---- code/bngblaster/src/bgp/bgp_ctrl.c | 21 +++++++- code/bngblaster/src/bgp/bgp_def.h | 14 ++++-- code/bngblaster/src/bgp/bgp_receive.c | 58 ++++++++++----------- code/bngblaster/src/bgp/bgp_session.c | 59 +++++++++++++++++++++- 7 files changed, 188 insertions(+), 48 deletions(-) diff --git a/code/bngblaster/src/bbl_tcp.c b/code/bngblaster/src/bbl_tcp.c index 3b11f80b..4d4d1fee 100644 --- a/code/bngblaster/src/bbl_tcp.c +++ b/code/bngblaster/src/bbl_tcp.c @@ -1,6 +1,6 @@ /* - * BNG Blaster (BBL) - LwIP - * + * BNG Blaster (BBL) - TCP (lwIP) + * * Christian Giese, February 2022 * * Copyright (C) 2020-2021, RtBrick, Inc. @@ -115,6 +115,9 @@ bbl_tcp_sent_cb(void *arg, struct tcp_pcb *tpcb, u16_t len) { } else if(tcpc->pcb->unacked == NULL && tcpc->pcb->unsent == NULL) { /* Idle means that it is save to replace buffer. */ tcpc->state = BBL_TCP_STATE_IDLE; + if(tcpc->idle_cb) { + (tcpc->idle_cb)(tcpc->arg); + } } return ERR_OK; } diff --git a/code/bngblaster/src/bbl_tcp.h b/code/bngblaster/src/bbl_tcp.h index 4bf9ba05..57228240 100644 --- a/code/bngblaster/src/bbl_tcp.h +++ b/code/bngblaster/src/bbl_tcp.h @@ -24,7 +24,7 @@ typedef enum bbl_tcp_state_ { BBL_TCP_STATE_SENDING, } bbl_tcp_state_t; -typedef void (*bbl_tcp_connected_fn)(void *arg); +typedef void (*bbl_tcp_callback_fn)(void *arg); typedef void (*bbl_tcp_receive_fn)(void *arg, uint8_t *buf, uint16_t len); typedef void (*bbl_tcp_error_fn)(void *arg, err_t err); typedef err_t (*bbl_tcp_poll_fn)(void *arg, struct tcp_pcb *tpcb); @@ -45,7 +45,8 @@ typedef struct bbl_tcp_ctx_ struct tcp_pcb *pcb; - bbl_tcp_connected_fn connected_cb; /* application connected callback */ + bbl_tcp_callback_fn connected_cb; /* application connected callback */ + bbl_tcp_callback_fn idle_cb; /* application idle callback */ bbl_tcp_receive_fn receive_cb; /* application receive callback */ bbl_tcp_error_fn error_cb; /* application error callback */ diff --git a/code/bngblaster/src/bgp/bgp.c b/code/bngblaster/src/bgp/bgp.c index 5df50af5..57b170a7 100644 --- a/code/bngblaster/src/bgp/bgp.c +++ b/code/bngblaster/src/bgp/bgp.c @@ -9,11 +9,16 @@ #include "bgp.h" static bgp_raw_update_t * -bgp_raw_update_load(char *file) { - bgp_raw_update_t *raw_update; +bgp_raw_update_load(char *file, bool decode_file) { + bgp_raw_update_t *raw_update = NULL; long fsize; double fsize_kb; + uint8_t *buf = NULL; + uint32_t len = 0; + uint16_t msg_len; + uint8_t msg_type; + /* Open file */ FILE *f = fopen(file, "rb"); if (f == NULL) { @@ -28,19 +33,64 @@ bgp_raw_update_load(char *file) { fseek(f, 0, SEEK_SET); /* Load file into memory */ - LOG(INFO, "Load BGP RAW update file %s (%.2f KB)\n", file, fsize_kb); raw_update = malloc(sizeof(bgp_raw_update_t)); raw_update->file = file; raw_update->buf = malloc(fsize); raw_update->len = fsize; if(fread(raw_update->buf, fsize, 1, f) != 1) { + fclose(f); LOG(ERROR, "Failed to read BGP RAW update file %s\n", file); - free(raw_update->buf); - free(raw_update); - return NULL; + goto ERROR; } fclose(f); + + if(decode_file) { + /* Decode update stream */ + buf = raw_update->buf; + len = raw_update->len; + while(len) { + if(len < BGP_MIN_MESSAGE_SIZE) { + goto DECODE_ERROR; + } + if(*(uint64_t*)buf != UINT64_MAX) { + goto DECODE_ERROR; + } + BUMP_BUFFER(buf, len, sizeof(uint64_t)); + if(*(uint64_t*)buf != UINT64_MAX) { + goto DECODE_ERROR; + } + BUMP_BUFFER(buf, len, sizeof(uint64_t)); + msg_len = be16toh(*(uint16_t*)buf); + BUMP_BUFFER(buf, len, sizeof(uint16_t)); + msg_type = *buf; + BUMP_BUFFER(buf, len, sizeof(uint8_t)); + if(msg_len < BGP_MIN_MESSAGE_SIZE || + msg_len > BGP_MAX_MESSAGE_SIZE) { + goto DECODE_ERROR; + } + if((msg_len - BGP_MIN_MESSAGE_SIZE) > len) { + goto DECODE_ERROR; + } + if(msg_type == BGP_MSG_UPDATE) { + raw_update->updates++; + } + BUMP_BUFFER(buf, len, (msg_len - BGP_MIN_MESSAGE_SIZE)); + } + } + LOG(INFO, "Loaded BGP RAW update file %s (%.2f KB, %u updates)\n", + file, fsize_kb, raw_update->updates); return raw_update; + +DECODE_ERROR: + LOG(ERROR, "Failed to decode BGP RAW update file %s\n", file); +ERROR: + if(raw_update) { + if(raw_update->buf) { + free(raw_update->buf); + } + free(raw_update); + } + return NULL; } /** @@ -106,9 +156,13 @@ bgp_init(bbl_ctx_s *ctx) { raw_update = raw_update->next; } if(!session->raw_update) { - session->raw_update = bgp_raw_update_load(config->raw_update_file); - session->raw_update->next = raw_update_root; - raw_update_root = session->raw_update; + session->raw_update = bgp_raw_update_load(config->raw_update_file, true); + if(session->raw_update) { + session->raw_update->next = raw_update_root; + raw_update_root = session->raw_update; + } else { + return false; + } } } diff --git a/code/bngblaster/src/bgp/bgp_ctrl.c b/code/bngblaster/src/bgp/bgp_ctrl.c index eec5231f..28cdcb26 100644 --- a/code/bngblaster/src/bgp/bgp_ctrl.c +++ b/code/bngblaster/src/bgp/bgp_ctrl.c @@ -11,12 +11,25 @@ json_t * bgp_ctrl_session(bgp_session_t *session) { json_t *root = NULL; + json_t *stats = NULL; if(!session) { return NULL; } - root = json_pack("{ss ss ss si si ss ss si si ss }", + stats = json_pack("{si si si si si si}", + "messages-rx", session->stats.message_rx, + "messages-tx", session->stats.message_tx, + "keepalive-rx", session->stats.keepalive_rx, + "keepalive-tx", session->stats.keepalive_tx, + "update-rx", session->stats.update_rx, + "update-tx", session->stats.update_tx); + + if(!stats) { + return NULL; + } + + root = json_pack("{ss ss ss si si ss ss si si ss so}", "interface", session->interface->name, "local-address", format_ipv4_address(&session->ipv4_local_address), "local-id", format_ipv4_address(&session->config->id), @@ -26,7 +39,11 @@ bgp_ctrl_session(bgp_session_t *session) { "peer-id", format_ipv4_address(&session->peer.id), "peer-as", session->peer.as, "peer-holdtime", session->peer.holdtime, - "state", bgp_session_state_string(session->state)); + "state", bgp_session_state_string(session->state), + "stats", stats); + if(!root) { + if(stats) json_decref(stats); + } return root; } \ No newline at end of file diff --git a/code/bngblaster/src/bgp/bgp_def.h b/code/bngblaster/src/bgp/bgp_def.h index 39fdd8ec..668a1224 100644 --- a/code/bngblaster/src/bgp/bgp_def.h +++ b/code/bngblaster/src/bgp/bgp_def.h @@ -12,8 +12,8 @@ /* DEFINITIONS ... */ #define BGP_PORT 179 -#define BGP_MIN_MESSAGE_SIZE 19 -#define BGP_MAX_MESSAGE_SIZE 4096 +#define BGP_MIN_MESSAGE_SIZE 19U +#define BGP_MAX_MESSAGE_SIZE 4096U #define BGP_BUF_SIZE 256*1024 #define BGP_DEFAULT_AS 65000 #define BGP_DEFAULT_HOLDTIME 90 @@ -43,6 +43,7 @@ typedef struct bgp_raw_update_ { uint8_t *buf; uint32_t len; + uint32_t updates; /* Pointer to next instance */ struct bgp_raw_update_ *next; @@ -97,7 +98,6 @@ typedef struct bgp_session_ { io_buffer_t write_buf; bgp_state_t state; - uint16_t keepalive_countdown; struct { uint32_t as; @@ -106,13 +106,21 @@ typedef struct bgp_session_ { } peer; struct { + uint32_t message_rx; + uint32_t message_tx; uint32_t keepalive_rx; + uint32_t keepalive_tx; uint32_t update_rx; + uint32_t update_tx; } stats; bgp_raw_update_t *raw_update; bool raw_update_send; + struct timespec established_timestamp; + struct timespec update_start_timestamp; + struct timespec update_stop_timestamp; + bool teardown; uint8_t error_code; uint8_t error_subcode; diff --git a/code/bngblaster/src/bgp/bgp_receive.c b/code/bngblaster/src/bgp/bgp_receive.c index bc9a00d0..14d21da3 100644 --- a/code/bngblaster/src/bgp/bgp_receive.c +++ b/code/bngblaster/src/bgp/bgp_receive.c @@ -69,10 +69,25 @@ struct keyval_ bgp_notification_cease_error_values[] = { { 0, NULL} }; +static void +bgp_decode_error(bgp_session_t *session) { + LOG(BGP, "BGP (%s %s - %s) invalid message received\n", + session->interface->name, + format_ipv4_address(&session->ipv4_local_address), + format_ipv4_address(&session->ipv4_peer_address)); + + if(!session->error_code) { + session->error_code = 1; /* Message header error */ + session->error_subcode = 2; /* Bad message length */ + } + bgp_session_close(session); +} + static void bgp_open(bgp_session_t *session, uint8_t *start, uint16_t length) { if(length < 28) { - goto ERROR; + bgp_decode_error(session); + return; } session->peer.as = read_be_uint(start+20, 2); session->peer.holdtime = read_be_uint(start+22, 2); @@ -86,18 +101,6 @@ bgp_open(bgp_session_t *session, uint8_t *start, uint16_t length) { bgp_session_state_change(session, BGP_OPENCONFIRM); return; - -ERROR: - LOG(BGP, "BGP (%s %s - %s) invalid open message received\n", - session->interface->name, - format_ipv4_address(&session->ipv4_local_address), - format_ipv4_address(&session->ipv4_peer_address)); - - if(!session->error_code) { - session->error_code = 1; /* Message header error */ - session->error_subcode = 2; /* Bad message length */ - } - bgp_session_close(session); } static void @@ -105,7 +108,8 @@ bgp_notification(bgp_session_t *session, uint8_t *start, uint16_t length) { uint8_t error_code, error_subcode; if(length < 21) { - goto ERROR; + bgp_decode_error(session); + return; } error_code = *(start+19); @@ -154,18 +158,6 @@ bgp_notification(bgp_session_t *session, uint8_t *start, uint16_t length) { session->error_code = 0; bgp_session_close(session); return; - -ERROR: - LOG(BGP, "BGP (%s %s - %s) invalid notification message received\n", - session->interface->name, - format_ipv4_address(&session->ipv4_local_address), - format_ipv4_address(&session->ipv4_peer_address)); - - if(!session->error_code) { - session->error_code = 1; /* Message header error */ - session->error_subcode = 2; /* Bad message length */ - } - bgp_session_close(session); } /* @@ -202,17 +194,25 @@ bpg_read(bgp_session_t *session) { size = buffer->idx - buffer->start_idx; /* Minimum message size */ - if (size < 19) { + if (size < BGP_MIN_MESSAGE_SIZE) { + break; + } + + length = read_be_uint(start+16, 2); + if (length < BGP_MIN_MESSAGE_SIZE || + length > BGP_MAX_MESSAGE_SIZE) { + bgp_decode_error(session); break; } /* Full message on the wire to consume? */ - length = read_be_uint(start+16, 2); - type = *(start+18); if (length > size) { break; } + type = *(start+18); + + session->stats.message_rx++; LOG(DEBUG, "BGP (%s %s - %s) read %s message\n", session->interface->name, format_ipv4_address(&session->ipv4_local_address), diff --git a/code/bngblaster/src/bgp/bgp_session.c b/code/bngblaster/src/bgp/bgp_session.c index 847875d7..cfd95f4f 100644 --- a/code/bngblaster/src/bgp/bgp_session.c +++ b/code/bngblaster/src/bgp/bgp_session.c @@ -75,11 +75,36 @@ bgp_session_keepalive_job(timer_s *timer) { if(session->tcpc && session->tcpc->state == BBL_TCP_STATE_IDLE) { bgp_session_reset_write_buffer(session); bgp_push_keepalive_message(session); - bgp_session_send(session); + if(bgp_session_send(session)) { + session->stats.message_tx++; + session->stats.keepalive_tx++; + } } } } +void +bgp_raw_update_stop_cb(void *arg) { + bgp_session_t *session = (bgp_session_t*)arg; + struct timespec time_diff; + + session->tcpc->idle_cb = NULL; + + clock_gettime(CLOCK_MONOTONIC, &session->update_stop_timestamp); + timespec_sub(&time_diff, + &session->update_stop_timestamp, + &session->update_start_timestamp); + + session->stats.message_tx += session->raw_update->updates; + session->stats.update_tx += session->raw_update->updates; + + LOG(BGP, "BGP (%s %s - %s) raw update stop after %lds\n", + session->interface->name, + format_ipv4_address(&session->ipv4_local_address), + format_ipv4_address(&session->ipv4_peer_address), + time_diff.tv_sec); +} + void bgp_session_update_job(timer_s *timer) { bgp_session_t *session = timer->data; @@ -89,6 +114,15 @@ bgp_session_update_job(timer_s *timer) { if(session->raw_update && !session->raw_update_send) { if(bbl_tcp_send(session->tcpc, session->raw_update->buf, session->raw_update->len)) { session->raw_update_send = true; + + LOG(BGP, "BGP (%s %s - %s) raw update start\n", + session->interface->name, + format_ipv4_address(&session->ipv4_local_address), + format_ipv4_address(&session->ipv4_peer_address)); + + clock_gettime(CLOCK_MONOTONIC, &session->update_start_timestamp); + session->tcpc->idle_cb = bgp_raw_update_stop_cb; + } else { goto RETRY; } @@ -110,6 +144,7 @@ bgp_session_state_opensent(bgp_session_t *session) { bgp_session_reset_write_buffer(session); bgp_push_open_message(session); bgp_session_send(session); + session->stats.message_tx++; } static void @@ -117,6 +152,8 @@ bgp_session_state_openconfirm(bgp_session_t *session) { bgp_session_reset_write_buffer(session); bgp_push_keepalive_message(session); bgp_session_send(session); + session->stats.message_tx++; + session->stats.keepalive_tx++; } static void @@ -124,6 +161,8 @@ bgp_session_state_estbalished(bgp_session_t *session) { bbl_ctx_s *ctx = session->interface->ctx; time_t keepalive_interval; + clock_gettime(CLOCK_MONOTONIC, &session->established_timestamp); + /* Start BGP keepalive */ keepalive_interval = session->peer.holdtime/2U; if(!keepalive_interval) { @@ -256,16 +295,33 @@ bgp_session_connect(bgp_session_t *session, time_t delay) { if(session->state == BGP_CLOSED) { bbl_tcp_ctx_free(session->tcpc); session->tcpc = NULL; + bgp_session_reset_read_buffer(session); bgp_session_reset_write_buffer(session); + session->peer.as = 0; session->peer.id = 0; session->peer.holdtime = 0; + + session->stats.message_rx = 0; + session->stats.message_tx = 0; session->stats.keepalive_rx = 0; + session->stats.keepalive_tx = 0; session->stats.update_rx = 0; + session->stats.update_tx = 0; + session->raw_update_send = false; + + session->established_timestamp.tv_sec = 0; + session->established_timestamp.tv_nsec = 0; + session->update_start_timestamp.tv_sec = 0; + session->update_start_timestamp.tv_nsec = 0; + session->update_stop_timestamp.tv_sec = 0; + session->update_stop_timestamp.tv_nsec = 0; + session->error_code = 0; session->error_subcode = 0; + bgp_session_state_change(session, BGP_IDLE); timer_add(&ctx->timer_root, &session->connect_timer, @@ -317,6 +373,7 @@ bgp_session_close(bgp_session_t *session) { bgp_session_reset_write_buffer(session); bgp_push_notification_message(session); bgp_session_send(session); + session->stats.message_tx++; bgp_session_state_change(session, BGP_CLOSING); delay = 3; }