#include #include #include #include #include #include #include #include #include #include #include "utable.h" #include "cjson/cJSON.h" #include "yyjson/yyjson.h" #include "uthash/utarray.h" #include "uthash/uthash.h" #define IPFIX_DEFUALT_VERSION 10 #define GEEDGE_NETWORKS_PEN_NUMBER 54450 #define IPFIX_DEFUALT_PEN_NUMBER GEEDGE_NETWORKS_PEN_NUMBER #define IPFIX_NONE_PEN_NUMBER -1 #define IPFIX_TEMPLATE_SET_ID 2 #define IPFIX_BUFF_MAX_SIZE 2048 #define IPFIX_ELEMENT_MAX_LEN 32 #ifndef MIN #define MIN(a, b) (((a) < (b)) ? (a) : (b)) #endif enum ipfix_type { IPFIX_UNSIGNED8 = 0, IPFIX_UNSIGNED16, IPFIX_UNSIGNED32, IPFIX_UNSIGNED64, IPFIX_VARIABLE_STRING, IPFIX_UNKNOWN }; struct ipfix_element { enum ipfix_type type; uint16_t element_id; uint16_t value_length; // if type is variable_string, value_length is 0 int PEN_number; char name[IPFIX_ELEMENT_MAX_LEN]; }; struct ipfix_template { uint16_t template_id; char *name; UT_array *elements_array; // utarray for current template elements }; struct ipfix_worker_context { int source_id; int sequence; size_t template_blob_length; char *template_blob; }; struct ipfix_exporter_schema { uint16_t n_worker; uint16_t version; int domain_id; int PEN_number; uint32_t templates_refresh_interval_s; UT_array *templates_array; // utarray for templates struct ipfix_worker_context *worker_context_array; // utarray for worker_context }; struct ipfix_message_head { uint16_t version; uint16_t length; int exporttime; int ipfix_message_sequence; int domain_id; }; #define IPFIX_MESSAGE_HEAD_LEN sizeof(struct ipfix_message_head) // from maat int load_file_to_memory(const char *file_name, unsigned char **pp_out, size_t *out_sz) { int ret = 0; FILE *fp = NULL; struct stat fstat_buf; size_t read_size = 0; ret = stat(file_name, &fstat_buf); if (ret != 0) { return -1; } fp = fopen(file_name, "r"); if (fp == NULL) { return -1; } *out_sz = fstat_buf.st_size; *pp_out = (unsigned char *)calloc(*out_sz + 1, sizeof(unsigned char)); read_size = fread(*pp_out, 1, *out_sz, fp); if (read_size != *out_sz) { free(*pp_out); *pp_out = NULL; } fclose(fp); fp = NULL; return 0; } int ipfix_exporter_get_type(struct ipfix_element *p_element, char *element_type) { if (strcmp(element_type, "string") == 0) { p_element->type = IPFIX_VARIABLE_STRING; p_element->value_length = 0; return 0; } else if (strcmp(element_type, "unsigned8") == 0) { p_element->type = IPFIX_UNSIGNED8; p_element->value_length = sizeof(uint8_t); return 0; } else if (strcmp(element_type, "unsigned16") == 0) { p_element->type = IPFIX_UNSIGNED16; p_element->value_length = sizeof(uint16_t); return 0; } else if (strcmp(element_type, "unsigned32") == 0) { p_element->type = IPFIX_UNSIGNED32; p_element->value_length = sizeof(uint32_t); return 0; } else if (strcmp(element_type, "unsigned64") == 0) { p_element->type = IPFIX_UNSIGNED64; p_element->value_length = sizeof(uint64_t); return 0; } else { return -1; } } int ipfix_template_payload_init(UT_array *templates_array, char **blob, size_t *blob_len) { if (templates_array == NULL || utarray_len(templates_array) == 0 || blob == NULL || blob_len == NULL) { return -1; } size_t n_templates = utarray_len(templates_array); size_t template_blob_len = 0; for (size_t i = 0; i < n_templates; i++) { struct ipfix_template *template_item = (struct ipfix_template *)utarray_eltptr(templates_array, i); size_t n_elements = utarray_len(template_item->elements_array); template_blob_len += (sizeof(uint16_t) + sizeof(uint16_t) + n_elements * (sizeof(uint16_t) + sizeof(uint16_t) + sizeof(uint32_t))); // template_id + field_count + element_id + length + PEN_number } template_blob_len += ((sizeof(uint16_t) + sizeof(uint16_t)) + IPFIX_MESSAGE_HEAD_LEN); // set_id + set_length + ipfix_message_head char *payload = (char *)calloc(template_blob_len, sizeof(char)); int offset = 0; offset += IPFIX_MESSAGE_HEAD_LEN; // skip ipfix_message_head *(uint16_t *)(payload + offset) = htons(IPFIX_TEMPLATE_SET_ID); // set_id offset += sizeof(uint16_t); // skip set_id *(uint16_t *)(payload + offset) = htons(template_blob_len - IPFIX_MESSAGE_HEAD_LEN); // set_length offset += sizeof(uint16_t); // skip set_length for (size_t i = 0; i < n_templates; i++) { struct ipfix_template *template_item = (struct ipfix_template *)utarray_eltptr(templates_array, i); size_t n_elements = utarray_len(template_item->elements_array); *(uint16_t *)(payload + offset) = htons(template_item->template_id); // template_id offset += sizeof(uint16_t); // skip template_id *(uint16_t *)(payload + offset) = htons(n_elements); // field_count offset += sizeof(uint16_t); // skip field_count for (size_t j = 0; j < n_elements; j++) { struct ipfix_element *p_element = (struct ipfix_element *)utarray_eltptr(template_item->elements_array, j); // set element_id if (p_element->PEN_number != IPFIX_NONE_PEN_NUMBER) { *(uint16_t *)(payload + offset) = htons(p_element->element_id | 0x8000); // Pen_provided is yes } else { *(uint16_t *)(payload + offset) = htons(p_element->element_id); // Pen_provided is no } offset += sizeof(uint16_t); // skip element_id // set element_length if (p_element->type == IPFIX_VARIABLE_STRING) { *(uint16_t *)(payload + offset) = htons(65535); // if element_length is 65535, it means variable length } else { *(uint16_t *)(payload + offset) = htons(p_element->value_length); // length } offset += sizeof(uint16_t); // skip length // set PEN_number if (p_element->PEN_number != IPFIX_NONE_PEN_NUMBER) { *(uint32_t *)(payload + offset) = htonl(p_element->PEN_number); // PEN_number offset += sizeof(uint32_t); // skip PEN_number } } } if (offset != (int)template_blob_len) { free(payload); payload = NULL; return -1; } *blob = payload; *blob_len = offset; return 0; } void ipfix_template_item_destroy(void *elt) { struct ipfix_template *template_item = (struct ipfix_template *)elt; if (template_item == NULL) { return; } if (template_item->elements_array != NULL) { utarray_free(template_item->elements_array); } if (template_item->name != NULL) { free(template_item->name); template_item->name = NULL; } return; } void ipfix_utarray_init(UT_array **_array, size_t sz_item, void (*dtor)(void *)) { if (_array == NULL || *_array != NULL) { return; } UT_icd icd = {0}; icd.sz = sz_item; icd.dtor = dtor; utarray_new(*_array, &icd); return; } UT_array *ipfix_elements_array_init(cJSON *root, const char *element_key, int PEN_number_global) { if (root == NULL || element_key == NULL) { return NULL; } cJSON *elements = cJSON_GetObjectItem(root, element_key); if (elements == NULL) { return NULL; } UT_array *elements_array = NULL; ipfix_utarray_init(&elements_array, sizeof(struct ipfix_element), NULL); size_t n_elements = cJSON_GetArraySize(elements); for (size_t i = 0; i < n_elements; i++) { struct ipfix_element element_item = {0}; cJSON *element = cJSON_GetArrayItem(elements, i); cJSON *element_id = cJSON_GetObjectItem(element, "element_id"); if (element_id == NULL || element_id->valueint == 0) // element_id不存在、类型不为Number、值为0 { goto error; } element_item.element_id = element_id->valueint; cJSON *element_type = cJSON_GetObjectItem(element, "element_type"); if (element_type == NULL || element_type->valuestring == NULL) // element_type不存在、类型不为String、值为NULL { goto error; } if (ipfix_exporter_get_type(&element_item, element_type->valuestring) == -1) // element_type不在支持范围内 { goto error; } cJSON *PEN_number = cJSON_GetObjectItem(element, "PEN_number"); if (PEN_number == NULL) // PEN_number不存在、类型不为Number { element_item.PEN_number = PEN_number_global; } else { element_item.PEN_number = PEN_number->valueint; } cJSON *element_name = cJSON_GetObjectItem(element, "element_name"); if (element_name == NULL || element_name->valuestring == NULL) // element_name不存在、类型不为String、值为NULL { goto error; } memcpy(element_item.name, element_name->valuestring, MIN(strlen(element_name->valuestring), IPFIX_ELEMENT_MAX_LEN - 1)); utarray_push_back(elements_array, &element_item); } return elements_array; error: utarray_free(elements_array); elements_array = NULL; return NULL; } void utable_ipfix_exporter_schema_free(struct ipfix_exporter_schema *instance) { if (instance == NULL) { return; } if (instance->templates_array) { utarray_free(instance->templates_array); instance->templates_array = NULL; } if (instance->worker_context_array) { for (size_t i = 0; i < instance->n_worker; i++) { if (instance->worker_context_array[i].template_blob) { free(instance->worker_context_array[i].template_blob); instance->worker_context_array[i].template_blob = NULL; } } } free(instance->worker_context_array); instance->worker_context_array = NULL; free(instance); instance = NULL; return; } struct ipfix_elements_hash { char *name; UT_array *elements_array; UT_hash_handle hh; }; void ipfix_elements_array_hash_destroy(struct ipfix_elements_hash *elements_array_hash) { if (elements_array_hash == NULL) { return; } struct ipfix_elements_hash *elements_array_item = NULL; struct ipfix_elements_hash *tmp = NULL; HASH_ITER(hh, elements_array_hash, elements_array_item, tmp) { HASH_DEL(elements_array_hash, elements_array_item); if (elements_array_item->name != NULL) { free(elements_array_item->name); elements_array_item->name = NULL; } if (elements_array_item->elements_array != NULL) { utarray_free(elements_array_item->elements_array); elements_array_item->elements_array = NULL; } free(elements_array_item); elements_array_item = NULL; } return; } struct ipfix_exporter_schema *utable_ipfix_exporter_schema_new(const char *ipfix_schema_json_path, uint16_t domain_id, uint16_t n_worker) { if (ipfix_schema_json_path == NULL) { return NULL; } size_t json_size = 0; unsigned char *json_str = NULL; int ret = load_file_to_memory(ipfix_schema_json_path, &json_str, &json_size); if (ret == -1) { return NULL; } cJSON *root = NULL; root = cJSON_Parse((const char *)json_str); if (root == NULL) { free(json_str); json_str = NULL; return NULL; } int version = 0; version = cJSON_GetObjectItem(root, "version")->valueint; if (version < IPFIX_DEFUALT_VERSION) { version = IPFIX_DEFUALT_VERSION; } int PEN_number = 0; PEN_number = cJSON_GetObjectItem(root, "PEN_number")->valueint; if (PEN_number <= 0) { PEN_number = IPFIX_DEFUALT_PEN_NUMBER; } uint32_t refresh_interval_s = 0; refresh_interval_s = cJSON_GetObjectItem(root, "refresh_interval_s")->valueint; if ((int)refresh_interval_s <= 0) { refresh_interval_s = 60; } size_t n_template = 0; cJSON *templates = NULL; templates = cJSON_GetObjectItem(root, "templates"); if (templates == NULL || cJSON_GetArraySize(templates) == 0) { free(json_str); json_str = NULL; return NULL; } n_template = cJSON_GetArraySize(templates); UT_array *templates_array = NULL; ipfix_utarray_init(&templates_array, sizeof(struct ipfix_template), ipfix_template_item_destroy); struct ipfix_elements_hash *elements_array_hash = NULL; for (size_t i = 0; i < n_template; i++) { struct ipfix_template template_item = {0}; cJSON *template_obj = cJSON_GetArrayItem(templates, i); cJSON *template_id = cJSON_GetObjectItem(template_obj, "template_id"); if (template_id == NULL || template_id->valueint < 256 || template_id->valueint == 0xffff) { goto error; } template_item.template_id = template_id->valueint; cJSON *template_name = cJSON_GetObjectItem(template_obj, "template_name"); if (template_name == NULL || template_name->valuestring == NULL) { goto error; } template_item.name = (char *)calloc(strlen(template_name->valuestring) + 1, sizeof(char)); memcpy(template_item.name, template_name->valuestring, strlen(template_name->valuestring)); cJSON *elements_key_array = cJSON_GetObjectItem(template_obj, "elements"); if (elements_key_array == NULL) { goto error; } size_t n_elements_key = cJSON_GetArraySize(elements_key_array); ipfix_utarray_init(&template_item.elements_array, sizeof(struct ipfix_element), NULL); for (size_t j = 0; j < n_elements_key; j++) { cJSON *element_key = cJSON_GetArrayItem(elements_key_array, j); struct ipfix_elements_hash *elements_array_item = NULL; HASH_FIND_STR(elements_array_hash, element_key->valuestring, elements_array_item); if (elements_array_item == NULL) { elements_array_item = (struct ipfix_elements_hash *)calloc(1, sizeof(struct ipfix_elements_hash)); elements_array_item->name = (char *)calloc(strlen(element_key->valuestring) + 1, sizeof(char)); memcpy(elements_array_item->name, element_key->valuestring, strlen(element_key->valuestring)); HASH_ADD_STR(elements_array_hash, name, elements_array_item); elements_array_item->elements_array = ipfix_elements_array_init(root, element_key->valuestring, PEN_number); } if (elements_array_item->elements_array == NULL) { goto error; } utarray_concat(template_item.elements_array, elements_array_item->elements_array); // merge elements_array } utarray_push_back(templates_array, &template_item); } ipfix_elements_array_hash_destroy(elements_array_hash); size_t sz_blob = 0; char *template_blob = NULL; ret = ipfix_template_payload_init(templates_array, &template_blob, &sz_blob); if (ret == -1 || sz_blob == 0 || template_blob == NULL) { goto error; } struct ipfix_worker_context *worker_context_array = (struct ipfix_worker_context *)calloc(n_worker, sizeof(struct ipfix_worker_context)); for (size_t i = 0; i < n_worker; i++) { worker_context_array[i].source_id = (domain_id << 16) | i; worker_context_array[i].sequence = 0; worker_context_array[i].template_blob_length = sz_blob; worker_context_array[i].template_blob = (char *)calloc(sz_blob, sizeof(char)); memcpy(worker_context_array[i].template_blob, template_blob, sz_blob); struct ipfix_message_head *p_message_head = (struct ipfix_message_head *)(worker_context_array[i].template_blob); p_message_head->version = htons(version); p_message_head->length = htons(sz_blob); p_message_head->domain_id = htonl(worker_context_array[i].source_id); } struct ipfix_exporter_schema *_instance = (struct ipfix_exporter_schema *)calloc(1, sizeof(struct ipfix_exporter_schema)); _instance->version = version; _instance->PEN_number = PEN_number; _instance->templates_refresh_interval_s = refresh_interval_s; _instance->domain_id = domain_id; _instance->n_worker = n_worker; _instance->templates_array = templates_array; _instance->worker_context_array = worker_context_array; free(template_blob); cJSON_Delete(root); free(json_str); json_str = NULL; return _instance; error: utarray_free(templates_array); cJSON_Delete(root); free(json_str); json_str = NULL; return NULL; } int utable_ipfix_template_get(struct ipfix_exporter_schema *instance, const char *template_name) { if (instance == NULL || template_name == NULL) { return -1; } size_t n_template = utarray_len(instance->templates_array); for (size_t i = 0; i < n_template; i++) { struct ipfix_template *template_item = (struct ipfix_template *)utarray_eltptr(instance->templates_array, i); if (strcmp(template_item->name, template_name) == 0) { return i; } } return -1; } char *ipfix_data_interger_serialize(char *buff, size_t *offset, size_t *buff_sz, int64_t value, size_t sz_value) { if (*offset + sz_value >= *buff_sz) { char *new_buff = (char *)realloc(buff, *offset + sz_value + IPFIX_BUFF_MAX_SIZE); if (new_buff == NULL) { free(buff); return NULL; } buff = new_buff; *buff_sz = *offset + sz_value + IPFIX_BUFF_MAX_SIZE; } switch (sz_value) { case 1: *(uint8_t *)(buff + *offset) = (uint8_t)value; *offset += 1; break; case 2: *(uint16_t *)(buff + *offset) = htons((uint16_t)value); *offset += 2; break; case 4: *(uint32_t *)(buff + *offset) = htonl((uint32_t)value); *offset += 4; break; case 8: *(uint64_t *)(buff + *offset) = htobe64((uint64_t)value); *offset += 8; break; } return buff; } char *ipfix_data_variable_serialize(char *buff, size_t *offset, size_t *buff_sz, char *value, size_t sz_value) { if (*offset + sz_value + 3 >= *buff_sz) // 3 means variable_string length > 255, need 3 bytes to describe length { char *new_buff = (char *)realloc(buff, *offset + sz_value + IPFIX_BUFF_MAX_SIZE); if (new_buff == NULL) { free(buff); return NULL; } buff = new_buff; *buff_sz = *offset + sz_value + IPFIX_BUFF_MAX_SIZE; } if (sz_value == 0 || value == NULL) // value == NULL need 1 byte to describe length { *(uint8_t *)(buff + *offset) = 0; *offset += 1; } else if (sz_value < 255) { *(uint8_t *)(buff + *offset) = (uint8_t)sz_value; *offset += 1; memcpy(buff + *offset, value, sz_value); *offset += sz_value; } else // >= 255 { *(uint8_t *)(buff + *offset) = 255; *offset += 1; *(uint16_t *)(buff + *offset) = htons(sz_value); *offset += 2; memcpy(buff + *offset, value, sz_value); *offset += sz_value; } return buff; } int utable_ipfix_data_flow_exporter(const struct utable *table, struct ipfix_exporter_schema *instance, int template_id, uint16_t worker_id, char **blob, size_t *blob_len) { if (table == NULL || instance == NULL || worker_id >= instance->n_worker || template_id < 0 || template_id >= (int)utarray_len(instance->templates_array)) { return -1; } struct ipfix_template *template_item = (struct ipfix_template *)utarray_eltptr(instance->templates_array, (unsigned int)template_id); if (template_item == NULL) { return -1; } size_t offset = 0; size_t n_elements = utarray_len(template_item->elements_array); size_t buff_sz = IPFIX_BUFF_MAX_SIZE; char *buff = (char *)calloc(IPFIX_BUFF_MAX_SIZE, sizeof(char)); offset += IPFIX_MESSAGE_HEAD_LEN; // skip ipfix_message_head *(uint16_t *)(buff + offset) = htons(template_item->template_id); // data_flow_set_id offset += sizeof(uint16_t); // skip data_flow_set_id offset += sizeof(uint16_t); // skip data_flow_set_length for (size_t i = 0; i < n_elements; i++) { struct ipfix_element *p_element = (struct ipfix_element *)utarray_eltptr(template_item->elements_array, i); enum utable_value_type value_type = utable_get_value_type(table, p_element->name); switch (value_type) { case utable_value_type_cstring: case utable_value_type_blob: { char *value = NULL; size_t sz_value = 0; utable_get0_cstring_value(table, p_element->name, &value, &sz_value); buff = ipfix_data_variable_serialize(buff, &offset, &buff_sz, value, sz_value); break; } case utable_value_type_integer: { int64_t value = 0; utable_get0_integer_value(table, p_element->name, &value); buff = ipfix_data_interger_serialize(buff, &offset, &buff_sz, value, p_element->value_length); break; } case utable_value_type_integer_array: { size_t n_array = 0; int64_t *value_array = NULL; utable_get0_integer_value_array(table, p_element->name, &value_array, &n_array); yyjson_mut_doc *doc = yyjson_mut_doc_new(NULL); yyjson_mut_val *arr = yyjson_mut_arr_with_sint64(doc, value_array, n_array); yyjson_mut_doc_set_root(doc, arr); char *value_array_str = yyjson_mut_write(doc, 0, NULL); if(value_array_str) { buff = ipfix_data_variable_serialize(buff, &offset, &buff_sz, value_array_str, strlen(value_array_str)); yyjson_mut_doc_free(doc); free(value_array_str); } break; } case utable_value_type_cstring_array: { size_t n_array = 0; char **value_array = NULL; size_t *value_sz = NULL; utable_get0_cstring_value_array(table, p_element->name, &value_array, &value_sz, &n_array); yyjson_mut_doc *doc = yyjson_mut_doc_new(NULL); yyjson_mut_val *arr = yyjson_mut_arr_with_strn(doc, (const char **)value_array, value_sz,n_array); yyjson_mut_doc_set_root(doc, arr); char *value_array_str = yyjson_mut_write(doc, 0, NULL); if (value_array_str) { buff = ipfix_data_variable_serialize(buff, &offset, &buff_sz, value_array_str, strlen(value_array_str)); yyjson_mut_doc_free(doc); free(value_array_str); } break; } default: // utable_value_type_undefined, ipfix append 0 { switch (p_element->type) { case IPFIX_UNSIGNED8: case IPFIX_UNSIGNED16: case IPFIX_UNSIGNED32: case IPFIX_UNSIGNED64: buff = ipfix_data_interger_serialize(buff, &offset, &buff_sz, 0, p_element->value_length); break; case IPFIX_VARIABLE_STRING: buff = ipfix_data_variable_serialize(buff, &offset, &buff_sz, NULL, 0); break; default: { free(buff); buff = NULL; return -1; } } break; } } } *(uint16_t *)(buff + IPFIX_MESSAGE_HEAD_LEN + sizeof(uint16_t)) = htons(offset - IPFIX_MESSAGE_HEAD_LEN); // data_flow_set_length struct ipfix_message_head *p_message_head = (struct ipfix_message_head *)(buff); p_message_head->version = htons(instance->version); p_message_head->length = htons(offset); p_message_head->exporttime = htonl(time(NULL)); // time_t is long int, but exporttime is int p_message_head->ipfix_message_sequence = htonl(instance->worker_context_array[worker_id].sequence++); p_message_head->domain_id = htonl(instance->worker_context_array[worker_id].source_id); *blob = buff; *blob_len = offset; return 0; } const char *utable_ipfix_template_flow_get0(struct ipfix_exporter_schema *instance, uint16_t worker_id, size_t *blob_len) { if (instance == NULL || instance->worker_context_array == NULL || worker_id >= instance->n_worker) { return NULL; } struct ipfix_message_head *p_message_head = (struct ipfix_message_head *)(instance->worker_context_array[worker_id].template_blob); p_message_head->exporttime = htonl(time(NULL)); // time_t is long int, but exporttime is int p_message_head->ipfix_message_sequence = htonl(instance->worker_context_array[worker_id].sequence); // template sequence is not increase *blob_len = instance->worker_context_array[worker_id].template_blob_length; return instance->worker_context_array[worker_id].template_blob; } uint32_t utable_ipfix_template_flow_refresh_interval_s(struct ipfix_exporter_schema *ipfix_schema) { return ipfix_schema->templates_refresh_interval_s; }