| 1 | // SPDX-License-Identifier: GPL-3.0-or-later |
| 2 | |
| 3 | #include "aclk_tx_msgs.h" |
| 4 | #include "aclk_util.h" |
| 5 | #include "aclk.h" |
| 6 | #include "aclk_capas.h" |
| 7 | |
| 8 | #include "schema-wrappers/proto_2_json.h" |
| 9 | |
| 10 | #ifndef __GNUC__ |
| 11 | #pragma region aclk_tx_msgs helper functions |
| 12 | #endif |
| 13 | |
| 14 | static void freez_aclk_publish_msg(void *ptr) { |
| 15 | freez(ptr); |
| 16 | } |
| 17 | |
| 18 | static bool aclk_size_add_overflow(size_t *total, size_t add) |
| 19 | { |
| 20 | if (add > SIZE_MAX - *total) |
| 21 | return true; |
| 22 | |
| 23 | *total += add; |
| 24 | return false; |
| 25 | } |
| 26 | |
| 27 | #define ACLK_HEADER_VERSION (2) |
| 28 | |
| 29 | uint16_t aclk_send_bin_message_subtopic_pid(mqtt_wss_client client, char *msg, size_t msg_len, enum aclk_topics subtopic, const char *msgname) |
| 30 | { |
| 31 | #ifndef ACLK_LOG_CONVERSATION_DIR |
| 32 | UNUSED(msgname); |
| 33 | #endif |
| 34 | uint16_t packet_id = 0; |
| 35 | const char *topic = aclk_get_topic(subtopic); |
| 36 | |
| 37 | if (unlikely(!topic)) { |
| 38 | netdata_log_error("Couldn't get topic. Aborting message send."); |
| 39 | return 0; |
| 40 | } |
| 41 | |
| 42 | if (aclklog_enabled) { |
| 43 | char *json = protomsg_to_json(msg, msg_len, msgname); |
| 44 | log_aclk_message_bin(json, strlen(json), 1, topic, msgname); |
| 45 | freez(json); |
| 46 | } |
| 47 | |
| 48 | int rc = mqtt_wss_publish5(client, (char *)topic, NULL, msg, &freez_aclk_publish_msg, msg_len, MQTT_WSS_PUB_QOS1, &packet_id); |
| 49 | if (rc != MQTT_WSS_OK) |
| 50 | packet_id = 0; |
| 51 | |
| 52 | return packet_id; |
| 53 | } |
| 54 | |
| 55 | #define V2_BIN_PAYLOAD_SEPARATOR "\x0D\x0A\x0D\x0A" |
| 56 | static short aclk_send_message_with_bin_payload(mqtt_wss_client client, json_object *msg, const char *topic, const void *payload, size_t payload_len) |
| 57 | { |
| 58 | uint16_t packet_id = 0; |
| 59 | const char *str; |
| 60 | char *full_msg = NULL; |
| 61 | size_t len; |
| 62 | |
| 63 | if (unlikely(!topic || topic[0] != '/')) { |
| 64 | netdata_log_error("Full topic required!"); |
| 65 | json_object_put(msg); |
| 66 | return HTTP_RESP_INTERNAL_SERVER_ERROR; |
| 67 | } |
| 68 | |
| 69 | str = json_object_to_json_string_ext(msg, JSON_C_TO_STRING_PLAIN); |
| 70 | len = strlen(str); |
| 71 | |
| 72 | size_t full_msg_len = len; |
| 73 | if (payload_len) |
| 74 | full_msg_len += strlen(V2_BIN_PAYLOAD_SEPARATOR) + payload_len; |
| 75 | |
| 76 | full_msg = mallocz(full_msg_len); |
| 77 | memcpy(full_msg, str, len); |
| 78 | json_object_put(msg); |
| 79 | |
| 80 | if (payload_len) { |
| 81 | memcpy(&full_msg[len], V2_BIN_PAYLOAD_SEPARATOR, sizeof(V2_BIN_PAYLOAD_SEPARATOR) - 1); |
| 82 | len += strlen(V2_BIN_PAYLOAD_SEPARATOR); |
| 83 | memcpy(&full_msg[len], payload, payload_len); |
| 84 | } |
| 85 | |
| 86 | int rc = mqtt_wss_publish5(client, (char*)topic, NULL, full_msg, &freez_aclk_publish_msg, full_msg_len, MQTT_WSS_PUB_QOS1, &packet_id); |
| 87 | |
| 88 | if (rc == MQTT_WSS_ERR_MSG_TOO_BIG) |
| 89 | return HTTP_RESP_CONTENT_TOO_LONG; |
| 90 | if (rc != MQTT_WSS_OK) |
| 91 | return HTTP_RESP_INTERNAL_SERVER_ERROR; |
| 92 | |
| 93 | return 0; |
| 94 | } |
| 95 | |
| 96 | /* |
| 97 | * Creates universal header common for all ACLK messages. User gets ownership of json object created. |
| 98 | * Usually this is freed by send function after message has been sent. |
| 99 | */ |
| 100 | static struct json_object *create_hdr(const char *type, const char *msg_id) |
| 101 | { |
| 102 | nd_uuid_t uuid; |
| 103 | char uuid_str[UUID_STR_LEN]; |
| 104 | json_object *tmp; |
| 105 | json_object *obj = json_object_new_object(); |
| 106 | time_t ts_secs; |
| 107 | usec_t ts_us; |
| 108 | |
| 109 | tmp = json_object_new_string(type); |
| 110 | json_object_object_add(obj, "type", tmp); |
| 111 | |
| 112 | if (unlikely(!msg_id)) { |
| 113 | uuid_generate(uuid); |
| 114 | uuid_unparse(uuid, uuid_str); |
| 115 | msg_id = uuid_str; |
| 116 | } |
| 117 | |
| 118 | ts_us = now_realtime_usec(); |
| 119 | ts_secs = ts_us / USEC_PER_SEC; |
| 120 | ts_us = ts_us % USEC_PER_SEC; |
| 121 | |
| 122 | tmp = json_object_new_string(msg_id); |
| 123 | json_object_object_add(obj, "msg-id", tmp); |
| 124 | |
| 125 | tmp = json_object_new_int64(ts_secs); |
| 126 | json_object_object_add(obj, "timestamp", tmp); |
| 127 | |
| 128 | // TODO handle this somehow on older json-c |
| 129 | // tmp = json_object_new_uint64(ts_us); |
| 130 | // probably jso->_to_json_string -> custom function |
| 131 | // jso->o.c_uint64 -> map this with pointer to signed int |
| 132 | // commit that implements json_object_new_uint64 is 3c3b592 |
| 133 | // between 0.14 and 0.15 |
| 134 | tmp = json_object_new_int64(ts_us); |
| 135 | json_object_object_add(obj, "timestamp-offset-usec", tmp); |
| 136 | |
| 137 | tmp = json_object_new_int64(aclk_session_sec); |
| 138 | json_object_object_add(obj, "connect", tmp); |
| 139 | |
| 140 | // TODO handle this somehow see above |
| 141 | // tmp = json_object_new_uint64(0 /* TODO aclk_session_us */); |
| 142 | tmp = json_object_new_int64(aclk_session_us); |
| 143 | json_object_object_add(obj, "connect-offset-usec", tmp); |
| 144 | |
| 145 | tmp = json_object_new_int(ACLK_HEADER_VERSION); |
| 146 | json_object_object_add(obj, "version", tmp); |
| 147 | |
| 148 | return obj; |
| 149 | } |
| 150 | |
| 151 | #ifndef __GNUC__ |
| 152 | #pragma endregion |
| 153 | #endif |
| 154 | |
| 155 | #ifndef __GNUC__ |
| 156 | #pragma region aclk_tx_msgs message generators |
| 157 | #endif |
| 158 | |
| 159 | void aclk_http_msg_v2_err(mqtt_wss_client client, const char *topic, const char *msg_id, int http_code, int ec, const char* emsg, const char *payload, size_t payload_len) |
| 160 | { |
| 161 | json_object *tmp, *msg; |
| 162 | msg = create_hdr("http", msg_id); |
| 163 | tmp = json_object_new_int(http_code); |
| 164 | json_object_object_add(msg, "http-code", tmp); |
| 165 | |
| 166 | tmp = json_object_new_int(ec); |
| 167 | json_object_object_add(msg, "error-code", tmp); |
| 168 | |
| 169 | tmp = json_object_new_string(emsg); |
| 170 | json_object_object_add(msg, "error-description", tmp); |
| 171 | |
| 172 | if (aclk_send_message_with_bin_payload(client, msg, topic, payload, payload_len)) { |
| 173 | netdata_log_error("Failed to send cancellation message for http reply %zu %s", payload_len, payload); |
| 174 | } |
| 175 | } |
| 176 | |
| 177 | short aclk_http_msg_v2(mqtt_wss_client client, const char *topic, const char *msg_id, usec_t t_exec, usec_t created, |
| 178 | short http_code, const char *payload, size_t payload_len) |
| 179 | { |
| 180 | json_object *tmp, *msg; |
| 181 | |
| 182 | msg = create_hdr("http", msg_id); |
| 183 | |
| 184 | tmp = json_object_new_int64(t_exec); |
| 185 | json_object_object_add(msg, "t-exec", tmp); |
| 186 | |
| 187 | tmp = json_object_new_int64(created); |
| 188 | json_object_object_add(msg, "t-rx", tmp); |
| 189 | |
| 190 | tmp = json_object_new_int(http_code); |
| 191 | json_object_object_add(msg, "http-code", tmp); |
| 192 | |
| 193 | short rc = aclk_send_message_with_bin_payload(client, msg, topic, payload, payload_len); |
| 194 | |
| 195 | switch (rc) { |
| 196 | case HTTP_RESP_CONTENT_TOO_LONG: |
| 197 | aclk_http_msg_v2_err(client, topic, msg_id, rc, CLOUD_EC_REQ_REPLY_TOO_BIG, CLOUD_EMSG_REQ_REPLY_TOO_BIG, NULL, 0); |
| 198 | break; |
| 199 | case HTTP_RESP_INTERNAL_SERVER_ERROR: |
| 200 | aclk_http_msg_v2_err(client, topic, msg_id, rc, CLOUD_EC_FAIL_TOPIC, CLOUD_EMSG_FAIL_TOPIC, payload, payload_len); |
| 201 | break; |
| 202 | default: |
| 203 | rc = http_code; |
| 204 | break; |
| 205 | } |
| 206 | return rc; |
| 207 | } |
| 208 | |
| 209 | short aclk_http_msg_v2_direct(mqtt_wss_client client, const char *topic, const char *msg_id, |
| 210 | usec_t t_exec, usec_t created, short http_code, |
| 211 | const char *http_headers, size_t http_headers_len, |
| 212 | const char *body, size_t body_len) |
| 213 | { |
| 214 | if (unlikely(!topic || topic[0] != '/')) { |
| 215 | netdata_log_error("Full topic required!"); |
| 216 | return HTTP_RESP_INTERNAL_SERVER_ERROR; |
| 217 | } |
| 218 | |
| 219 | // normalize NULL-with-len so the allocation size and the memcpys stay in sync |
| 220 | if (!http_headers) |
| 221 | http_headers_len = 0; |
| 222 | if (!body) |
| 223 | body_len = 0; |
| 224 | |
| 225 | json_object *msg = create_hdr("http", msg_id); |
| 226 | json_object *tmp; |
| 227 | |
| 228 | tmp = json_object_new_int64(t_exec); |
| 229 | json_object_object_add(msg, "t-exec", tmp); |
| 230 | |
| 231 | tmp = json_object_new_int64(created); |
| 232 | json_object_object_add(msg, "t-rx", tmp); |
| 233 | |
| 234 | tmp = json_object_new_int(http_code); |
| 235 | json_object_object_add(msg, "http-code", tmp); |
| 236 | |
| 237 | size_t json_len; |
| 238 | const char *json_str = json_object_to_json_string_length(msg, JSON_C_TO_STRING_PLAIN, &json_len); |
| 239 | |
| 240 | const size_t sep_len = sizeof(V2_BIN_PAYLOAD_SEPARATOR) - 1; |
| 241 | const bool has_payload = (http_headers_len != 0 || body_len != 0); |
| 242 | |
| 243 | size_t total = json_len; |
| 244 | if (unlikely( |
| 245 | (has_payload && aclk_size_add_overflow(&total, sep_len)) || |
| 246 | aclk_size_add_overflow(&total, http_headers_len) || |
| 247 | aclk_size_add_overflow(&total, body_len))) { |
| 248 | json_object_put(msg); |
| 249 | aclk_http_msg_v2_err(client, topic, msg_id, HTTP_RESP_CONTENT_TOO_LONG, CLOUD_EC_REQ_REPLY_TOO_BIG, CLOUD_EMSG_REQ_REPLY_TOO_BIG, NULL, 0); |
| 250 | return HTTP_RESP_CONTENT_TOO_LONG; |
| 251 | } |
| 252 | |
| 253 | char *raw = mallocz(total); |
| 254 | |
| 255 | size_t pos = 0; |
| 256 | memcpy(raw + pos, json_str, json_len); |
| 257 | pos += json_len; |
| 258 | json_object_put(msg); |
| 259 | |
| 260 | if (has_payload) { |
| 261 | memcpy(raw + pos, V2_BIN_PAYLOAD_SEPARATOR, sep_len); |
| 262 | pos += sep_len; |
| 263 | } |
| 264 | |
| 265 | if (http_headers_len) { |
| 266 | memcpy(raw + pos, http_headers, http_headers_len); |
| 267 | pos += http_headers_len; |
| 268 | } |
| 269 | |
| 270 | if (body_len) |
| 271 | memcpy(raw + pos, body, body_len); |
| 272 | |
| 273 | uint16_t packet_id = 0; |
| 274 | int rc = mqtt_wss_publish5(client, (char *)topic, NULL, raw, &freez_aclk_publish_msg, total, MQTT_WSS_PUB_QOS1, &packet_id); |
| 275 | |
| 276 | if (rc == MQTT_WSS_ERR_MSG_TOO_BIG) { |
| 277 | aclk_http_msg_v2_err(client, topic, msg_id, HTTP_RESP_CONTENT_TOO_LONG, CLOUD_EC_REQ_REPLY_TOO_BIG, CLOUD_EMSG_REQ_REPLY_TOO_BIG, NULL, 0); |
| 278 | return HTTP_RESP_CONTENT_TOO_LONG; |
| 279 | } |
| 280 | |
| 281 | if (rc != MQTT_WSS_OK) { |
| 282 | aclk_http_msg_v2_err(client, topic, msg_id, HTTP_RESP_INTERNAL_SERVER_ERROR, CLOUD_EC_SND_TIMEOUT, CLOUD_EMSG_SND_TIMEOUT, NULL, 0); |
| 283 | return HTTP_RESP_INTERNAL_SERVER_ERROR; |
| 284 | } |
| 285 | |
| 286 | return http_code; |
| 287 | } |
| 288 | |
| 289 | uint16_t aclk_send_agent_connection_update(mqtt_wss_client client, int reachable) { |
| 290 | size_t len; |
| 291 | uint16_t pid; |
| 292 | |
| 293 | update_agent_connection_t conn = { |
| 294 | .reachable = (reachable ? 1 : 0), |
| 295 | .lwt = 0, |
| 296 | .session_id = aclk_session_newarch, |
| 297 | .capabilities = aclk_get_agent_capas(), |
| 298 | }; |
| 299 | |
| 300 | CLAIM_ID claim_id = claim_id_get(); |
| 301 | if (unlikely(!claim_id_is_set(claim_id))) { |
| 302 | netdata_log_error("Internal error. Should not come here if not claimed"); |
| 303 | return 0; |
| 304 | } |
| 305 | |
| 306 | CLAIM_ID previous_claim_id = claim_id_get_last_working(); |
| 307 | if (claim_id_is_set(previous_claim_id)) |
| 308 | conn.claim_id = previous_claim_id.str; |
| 309 | else |
| 310 | conn.claim_id = claim_id.str; |
| 311 | |
| 312 | char *msg = generate_update_agent_connection(&len, &conn); |
| 313 | |
| 314 | if (!msg) { |
| 315 | netdata_log_error("Error generating agent::v1::UpdateAgentConnection payload"); |
| 316 | return 0; |
| 317 | } |
| 318 | |
| 319 | pid = aclk_send_bin_message_subtopic_pid(client, msg, len, ACLK_TOPICID_AGENT_CONN, "UpdateAgentConnection"); |
| 320 | if (claim_id_is_set(previous_claim_id)) |
| 321 | claim_id_clear_previous_working(); |
| 322 | |
| 323 | return pid; |
| 324 | } |
| 325 | |
| 326 | char *aclk_generate_lwt(size_t *size) { |
| 327 | update_agent_connection_t conn = { |
| 328 | .reachable = 0, |
| 329 | .lwt = 1, |
| 330 | .session_id = aclk_session_newarch, |
| 331 | .capabilities = NULL |
| 332 | }; |
| 333 | |
| 334 | CLAIM_ID claim_id = claim_id_get(); |
| 335 | if(!claim_id_is_set(claim_id)) { |
| 336 | netdata_log_error("Internal error. Should not come here if not claimed"); |
| 337 | return NULL; |
| 338 | } |
| 339 | conn.claim_id = claim_id.str; |
| 340 | |
| 341 | char *msg = generate_update_agent_connection(size, &conn); |
| 342 | |
| 343 | if (!msg) |
| 344 | netdata_log_error("Error generating agent::v1::UpdateAgentConnection payload for LWT"); |
| 345 | |
| 346 | return msg; |
| 347 | } |
| 348 | |
| 349 | #ifndef __GNUC__ |
| 350 | #pragma endregion |
| 351 | #endif |