src/aclk/aclk_util.c
// SPDX-License-Identifier: GPL-3.0-or-later
#include "aclk_util.h"
#ifdef ENABLE_ACLK
#include "aclk_proxy.h"
#include "daemon/common.h"
usec_t aclk_session_newarch = 0;
aclk_env_t *aclk_env = NULL;
int chart_batch_id;
aclk_encoding_type_t aclk_encoding_type_t_from_str(const char *str) {
if (!strcmp(str, "json")) {
return ACLK_ENC_JSON;
}
if (!strcmp(str, "proto")) {
return ACLK_ENC_PROTO;
}
return ACLK_ENC_UNKNOWN;
}
aclk_transport_type_t aclk_transport_type_t_from_str(const char *str) {
if (!strcmp(str, "MQTTv3")) {
return ACLK_TRP_MQTT_3_1_1;
}
if (!strcmp(str, "MQTTv5")) {
return ACLK_TRP_MQTT_5;
}
return ACLK_TRP_UNKNOWN;
}
void aclk_transport_desc_t_destroy(aclk_transport_desc_t *trp_desc) {
freez(trp_desc->endpoint);
}
void aclk_env_t_destroy(aclk_env_t *env) {
freez(env->auth_endpoint);
if (env->transports) {
for (size_t i = 0; i < env->transport_count; i++) {
if(env->transports[i]) {
aclk_transport_desc_t_destroy(env->transports[i]);
freez(env->transports[i]);
env->transports[i] = NULL;
}
}
freez(env->transports);
}
if (env->capabilities) {
for (size_t i = 0; i < env->capability_count; i++)
freez(env->capabilities[i]);
freez(env->capabilities);
}
}
int aclk_env_has_capa(const char *capa)
{
for (int i = 0; i < (int) aclk_env->capability_count; i++) {
if (!strcasecmp(capa, aclk_env->capabilities[i]))
return 1;
}
return 0;
}
#ifdef ACLK_LOG_CONVERSATION_DIR
volatile int aclk_conversation_log_counter = 0;
#endif
#define ACLK_TOPIC_PREFIX "/agent/"
struct aclk_topic {
enum aclk_topics topic_id;
// as received from cloud - we keep this for
// eventual topic list update when claim_id changes
char *topic_recvd;
// constructed topic
char *topic;
};
// This helps to cache finalized topics (assembled with claim_id)
// to not have to alloc or create buffer and construct topic every
// time message is sent as in old ACLK
static struct aclk_topic **aclk_topic_cache = NULL;
static size_t aclk_topic_cache_items = 0;
void free_topic_cache(void)
{
if (aclk_topic_cache) {
for (size_t i = 0; i < aclk_topic_cache_items; i++) {
freez(aclk_topic_cache[i]->topic);
freez(aclk_topic_cache[i]->topic_recvd);
freez(aclk_topic_cache[i]);
}
freez(aclk_topic_cache);
aclk_topic_cache = NULL;
aclk_topic_cache_items = 0;
}
}
#define JSON_TOPIC_KEY_TOPIC "topic"
#define JSON_TOPIC_KEY_NAME "name"
struct topic_name {
enum aclk_topics id;
// cloud name - how is it called
// in answer to /password endpoint
const char *name;
} topic_names[] = {
{ .id = ACLK_TOPICID_CHART, .name = "chart" },
{ .id = ACLK_TOPICID_ALARMS, .name = "alarms" },
{ .id = ACLK_TOPICID_METADATA, .name = "meta" },
{ .id = ACLK_TOPICID_COMMAND, .name = "inbox-cmd" },
{ .id = ACLK_TOPICID_AGENT_CONN, .name = "agent-connection" },
{ .id = ACLK_TOPICID_CMD_NG_V1, .name = "inbox-cmd-v1" },
{ .id = ACLK_TOPICID_CREATE_NODE, .name = "create-node-instance" },
{ .id = ACLK_TOPICID_NODE_CONN, .name = "node-instance-connection" },
{ .id = ACLK_TOPICID_CHART_DIMS, .name = "chart-and-dims-updated" },
{ .id = ACLK_TOPICID_CHART_CONFIGS_UPDATED, .name = "chart-configs-updated" },
{ .id = ACLK_TOPICID_CHART_RESET, .name = "reset-charts" },
{ .id = ACLK_TOPICID_RETENTION_UPDATED, .name = "chart-retention-updated" },
{ .id = ACLK_TOPICID_NODE_INFO, .name = "node-instance-info" },
{ .id = ACLK_TOPICID_ALARM_LOG, .name = "alarm-log-v2" },
{ .id = ACLK_TOPICID_ALARM_CHECKPOINT, .name = "alarm-checkpoint" },
{ .id = ACLK_TOPICID_ALARM_CONFIG, .name = "alarm-config" },
{ .id = ACLK_TOPICID_ALARM_SNAPSHOT, .name = "alarm-snapshot-v2" },
{ .id = ACLK_TOPICID_NODE_COLLECTORS, .name = "node-instance-collectors" },
{ .id = ACLK_TOPICID_CTXS_SNAPSHOT, .name = "contexts-snapshot" },
{ .id = ACLK_TOPICID_CTXS_UPDATED, .name = "contexts-updated" },
{ .id = ACLK_TOPICID_UNKNOWN, .name = NULL }
};
enum aclk_topics compulsory_topics[] = {
// TODO remove old topics once not needed anymore
ACLK_TOPICID_CHART, //TODO from legacy
ACLK_TOPICID_ALARMS, //TODO from legacy
ACLK_TOPICID_METADATA, //TODO from legacy
ACLK_TOPICID_COMMAND,
ACLK_TOPICID_AGENT_CONN,
ACLK_TOPICID_CMD_NG_V1,
ACLK_TOPICID_CREATE_NODE,
ACLK_TOPICID_NODE_CONN,
ACLK_TOPICID_CHART_DIMS,
ACLK_TOPICID_CHART_CONFIGS_UPDATED,
ACLK_TOPICID_CHART_RESET,
ACLK_TOPICID_RETENTION_UPDATED,
ACLK_TOPICID_NODE_INFO,
ACLK_TOPICID_ALARM_LOG,
ACLK_TOPICID_ALARM_CHECKPOINT,
ACLK_TOPICID_ALARM_CONFIG,
ACLK_TOPICID_ALARM_SNAPSHOT,
ACLK_TOPICID_NODE_COLLECTORS,
ACLK_TOPICID_CTXS_SNAPSHOT,
ACLK_TOPICID_CTXS_UPDATED,
ACLK_TOPICID_UNKNOWN
};
static enum aclk_topics topic_name_to_id(const char *name) {
struct topic_name *topic = topic_names;
while (topic->name) {
if (!strcmp(topic->name, name)) {
return topic->id;
}
topic++;
}
return ACLK_TOPICID_UNKNOWN;
}
static const char *topic_id_to_name(enum aclk_topics tid) {
struct topic_name *topic = topic_names;
while (topic->name) {
if (topic->id == tid)
return topic->name;
topic++;
}
return "unknown";
}
#define CLAIM_ID_REPLACE_TAG "#{claim_id}"
static void topic_generate_final(struct aclk_topic *t) {
char *dest;
char *replace_tag = strstr(t->topic_recvd, CLAIM_ID_REPLACE_TAG);
if (!replace_tag)
return;
rrdhost_aclk_state_lock(localhost);
if (unlikely(!localhost->aclk_state.claimed_id)) {
netdata_log_error("This should never be called if agent not claimed");
rrdhost_aclk_state_unlock(localhost);
return;
}
t->topic = mallocz(strlen(t->topic_recvd) + 1 - strlen(CLAIM_ID_REPLACE_TAG) + strlen(localhost->aclk_state.claimed_id));
memcpy(t->topic, t->topic_recvd, replace_tag - t->topic_recvd);
dest = t->topic + (replace_tag - t->topic_recvd);
memcpy(dest, localhost->aclk_state.claimed_id, strlen(localhost->aclk_state.claimed_id));
dest += strlen(localhost->aclk_state.claimed_id);
rrdhost_aclk_state_unlock(localhost);
replace_tag += strlen(CLAIM_ID_REPLACE_TAG);
strcpy(dest, replace_tag);
dest += strlen(replace_tag);
*dest = 0;
}
static int topic_cache_add_topic(struct json_object *json, struct aclk_topic *topic)
{
struct json_object_iterator it;
struct json_object_iterator itEnd;
it = json_object_iter_begin(json);
itEnd = json_object_iter_end(json);
while (!json_object_iter_equal(&it, &itEnd)) {
if (!strcmp(json_object_iter_peek_name(&it), JSON_TOPIC_KEY_NAME)) {
if (json_object_get_type(json_object_iter_peek_value(&it)) != json_type_string) {
netdata_log_error("topic dictionary key \"" JSON_TOPIC_KEY_NAME "\" is expected to be json_type_string");
return 1;
}
topic->topic_id = topic_name_to_id(json_object_get_string(json_object_iter_peek_value(&it)));
if (topic->topic_id == ACLK_TOPICID_UNKNOWN) {
netdata_log_debug(D_ACLK, "topic dictionary has unknown topic name \"%s\"", json_object_get_string(json_object_iter_peek_value(&it)));
}
json_object_iter_next(&it);
continue;
}
if (!strcmp(json_object_iter_peek_name(&it), JSON_TOPIC_KEY_TOPIC)) {
if (json_object_get_type(json_object_iter_peek_value(&it)) != json_type_string) {
netdata_log_error("topic dictionary key \"" JSON_TOPIC_KEY_TOPIC "\" is expected to be json_type_string");
return 1;
}
topic->topic_recvd = strdupz(json_object_get_string(json_object_iter_peek_value(&it)));
json_object_iter_next(&it);
continue;
}
netdata_log_error("topic dictionary has Unknown/Unexpected key \"%s\" in topic description. Ignoring!", json_object_iter_peek_name(&it));
json_object_iter_next(&it);
}
if (!topic->topic_recvd) {
netdata_log_error("topic dictionary Missig compulsory key %s", JSON_TOPIC_KEY_TOPIC);
return 1;
}
topic_generate_final(topic);
aclk_topic_cache_items++;
return 0;
}
int aclk_generate_topic_cache(struct json_object *json)
{
json_object *obj;
size_t array_size = json_object_array_length(json);
if (!array_size) {
netdata_log_error("Empty topic list!");
return 1;
}
if (aclk_topic_cache)
free_topic_cache();
aclk_topic_cache = callocz(array_size, sizeof(struct aclk_topic *));
for (size_t i = 0; i < array_size; i++) {
obj = json_object_array_get_idx(json, i);
if (json_object_get_type(obj) != json_type_object) {
netdata_log_error("expected json_type_object");
return 1;
}
aclk_topic_cache[i] = callocz(1, sizeof(struct aclk_topic));
if (topic_cache_add_topic(obj, aclk_topic_cache[i])) {
netdata_log_error("failed to parse topic @idx=%d", (int)i);
return 1;
}
}
for (int i = 0; compulsory_topics[i] != ACLK_TOPICID_UNKNOWN; i++) {
if (!aclk_get_topic(compulsory_topics[i])) {
netdata_log_error("missing compulsory topic \"%s\" in password response from cloud", topic_id_to_name(compulsory_topics[i]));
return 1;
}
}
return 0;
}
/*
* Build a topic based on sub_topic and final_topic
* if the sub topic starts with / assume that is an absolute topic
*
*/
const char *aclk_get_topic(enum aclk_topics topic)
{
if (!aclk_topic_cache) {
netdata_log_error("Topic cache not initialized");
return NULL;
}
for (size_t i = 0; i < aclk_topic_cache_items; i++) {
if (aclk_topic_cache[i]->topic_id == topic)
return aclk_topic_cache[i]->topic;
}
netdata_log_error("Unknown topic");
return NULL;
}
/*
* Allows iterating all topics in topic cache without
* having to resort to callbacks.
*/
const char *aclk_topic_cache_iterate(aclk_topic_cache_iter_t *iter)
{
if (!aclk_topic_cache) {
netdata_log_error("Topic cache not initialized when %s was called.", __FUNCTION__);
return NULL;
}
if (*iter >= aclk_topic_cache_items)
return NULL;
return aclk_topic_cache[(*iter)++]->topic;
}
/*
* TBEB with randomness
*
* @param reset 1 - to reset the delay,
* 0 - to advance a step and calculate sleep time in ms
* @param min, max in seconds
* @returns delay in ms
*
*/
unsigned long int aclk_tbeb_delay(int reset, int base, unsigned long int min, unsigned long int max) {
static int attempt = -1;
if (reset) {
attempt = -1;
return 0;
}
attempt++;
if (attempt == 0) {
srandom(time(NULL));
return 0;
}
unsigned long int delay = pow(base, attempt - 1);
delay *= MSEC_PER_SEC;
delay += (random() % (MAX(1000, delay/2)));
if (delay <= min * MSEC_PER_SEC)
return min;
if (delay >= max * MSEC_PER_SEC)
return max;
return delay;
}
static inline int aclk_parse_pair(const char *src, const char c, char **a, char **b)
{
const char *ptr = strchr(src, c);
if (ptr == NULL)
return 1;
// allow empty string
/* if (!*(ptr+1))
return 1;*/
*a = callocz(1, ptr - src + 1);
memcpy (*a, src, ptr - src);
*b = strdupz(ptr+1);
return 0;
}
#define HTTP_PROXY_PREFIX "http://"
void aclk_set_proxy(char **ohost, int *port, char **uname, char **pwd, enum mqtt_wss_proxy_type *type)
{
ACLK_PROXY_TYPE pt;
const char *ptr = aclk_get_proxy(&pt);
char *tmp;
if (pt != PROXY_TYPE_HTTP)
return;
*uname = NULL;
*pwd = NULL;
*port = 0;
char *proxy = strdupz(ptr);
ptr = proxy;
if (!strncmp(ptr, HTTP_PROXY_PREFIX, strlen(HTTP_PROXY_PREFIX)))
ptr += strlen(HTTP_PROXY_PREFIX);
if ((tmp = strchr(ptr, '@'))) {
*tmp = 0;
if(aclk_parse_pair(ptr, ':', uname, pwd)) {
error_report("Failed to get username and password for proxy. Will attempt connection without authentication");
}
ptr = tmp+1;
}
if (!*ptr) {
freez(proxy);
freez(*uname);
freez(*pwd);
return;
}
if ((tmp = strchr(ptr, ':'))) {
*tmp = 0;
tmp++;
if(*tmp)
*port = atoi(tmp);
}
*ohost = strdupz(ptr);
if (*port <= 0 || *port > 65535)
*port = 8080;
if (type)
*type = MQTT_WSS_PROXY_HTTP;
else {
freez(*uname);
freez(*pwd);
}
freez(proxy);
}
#endif /* ENABLE_ACLK */
#if defined(OPENSSL_VERSION_NUMBER) && OPENSSL_VERSION_NUMBER < OPENSSL_VERSION_110
static EVP_ENCODE_CTX *EVP_ENCODE_CTX_new(void)
{
EVP_ENCODE_CTX *ctx = OPENSSL_malloc(sizeof(*ctx));
if (ctx != NULL) {
memset(ctx, 0, sizeof(*ctx));
}
return ctx;
}
static void EVP_ENCODE_CTX_free(EVP_ENCODE_CTX *ctx)
{
OPENSSL_free(ctx);
return;
}
#endif
int base64_encode_helper(unsigned char *out, int *outl, const unsigned char *in, int in_len)
{
int len;
unsigned char *str = out;
EVP_ENCODE_CTX *ctx = EVP_ENCODE_CTX_new();
EVP_EncodeInit(ctx);
EVP_EncodeUpdate(ctx, str, outl, in, in_len);
str += *outl;
EVP_EncodeFinal(ctx, str, &len);
*outl += len;
str = out;
while(*str) {
if (*str != 0x0D && *str != 0x0A)
*out++ = *str++;
else
str++;
}
*out = 0;
EVP_ENCODE_CTX_free(ctx);
return 0;
}