implements ACLK env endpoint (#10833)
implements /env endpoint call and parsing of the response
Timotej S committed
Apr 19, 2021 at 17:52 UTC
a3c46ef3ec305bb2fa0e8f64e87e9ac8b99d6714
8 files changed
+654
-70
aclk/aclk.c
+90
-19
@@ -9,6 +9,7 @@
9
#include "aclk_util.h"
10
#include "aclk_rx_msgs.h"
11
#include "aclk_collector_list.h"
12
+#include "https_client.h"
13
14
#ifdef ACLK_LOG_CONVERSATION_DIR
15
#include <sys/types.h>
@@ -29,6 +30,8 @@ int aclk_pubacks_per_conn = 0; // How many PubAcks we got since MQTT conn est.
30
usec_t aclk_session_us = 0; // Used by the mqtt layer
31
time_t aclk_session_sec = 0; // Used by the mqtt layer
32
33
+aclk_env_t *aclk_env = NULL;
34
+
35
mqtt_wss_client mqttwss_client;
36
37
netdata_mutex_t aclk_shared_state_mutex = NETDATA_MUTEX_INITIALIZER;
@@ -138,8 +141,7 @@ static int wait_till_agent_claimed(void)
141
*/
142
static int wait_till_agent_claim_ready()
143
{
141
- int port;
142
- char *hostname = NULL;
144
+ url_t url;
145
while (!netdata_exit) {
146
if (wait_till_agent_claimed())
147
return 1;
@@ -154,15 +156,14 @@ static int wait_till_agent_claim_ready()
156
157
// We just check configuration is valid here
158
// TODO make it without malloc/free
157
- if (aclk_decode_base_url(cloud_base_url, &hostname, &port)) {
159
+ memset(&url, 0, sizeof(url_t));
160
+ if (url_parse(cloud_base_url, &url)) {
161
error("Agent is claimed but the configuration is invalid, please fix");
159
- freez(hostname);
160
- hostname = NULL;
162
+ url_t_destroy(&url);
163
sleep(5);
164
continue;
165
}
164
- freez(hostname);
165
- hostname = NULL;
166
+ url_t_destroy(&url);
167
168
if (!load_private_key()) {
169
sleep(5);
@@ -420,6 +421,23 @@ static int aclk_block_till_recon_allowed() {
421
return 0;
422
}
423
424
+#ifndef ACLK_DISABLE_CHALLENGE
425
+/* Cloud returns transport list ordered with highest
426
+ * priority first. This function selects highest prio
427
+ * transport that we can actually use (support)
428
+ */
429
+static int aclk_get_transport_idx(aclk_env_t *env) {
430
+ for (size_t i = 0; i < env->transport_count; i++) {
431
+ // currently we support only MQTT 3
432
+ // therefore select first transport that matches
433
+ if (env->transports[i]->type == ACLK_TRP_MQTT_3_1_1) {
434
+ return i;
435
+ }
436
+ }
437
+ return -1;
438
+}
439
+#endif
440
+
441
/* Attempts to make a connection to MQTT broker over WSS
442
* @param client instance of mqtt_wss_client
443
* @return 0 - Successfull Connection,
@@ -434,10 +452,14 @@ static int aclk_block_till_recon_allowed() {
452
#endif
453
static int aclk_attempt_to_connect(mqtt_wss_client client)
454
{
437
- char *aclk_hostname = NULL;
438
- int aclk_port;
455
+ int ret;
456
+
457
+ url_t base_url;
458
459
#ifndef ACLK_DISABLE_CHALLENGE
460
+ url_t auth_url;
461
+ url_t mqtt_url;
462
+
463
char *mqtt_otp_user = NULL;
464
char *mqtt_otp_pass = NULL;
465
#endif
@@ -455,9 +477,11 @@ static int aclk_attempt_to_connect(mqtt_wss_client client)
477
return 1;
478
479
info("Attempting connection now");
458
- if (aclk_decode_base_url(cloud_base_url, &aclk_hostname, &aclk_port)) {
480
+ memset(&base_url, 0, sizeof(url_t));
481
+ if (url_parse(cloud_base_url, &base_url)) {
482
error("ACLK base URL configuration key could not be parsed. Will retry in %d seconds.", CLOUD_BASE_URL_READ_RETRY);
483
sleep(CLOUD_BASE_URL_READ_RETRY);
484
+ url_t_destroy(&base_url);
485
continue;
486
}
487
@@ -473,29 +497,72 @@ static int aclk_attempt_to_connect(mqtt_wss_client client)
497
.will_flags = MQTT_WSS_PUB_QOS2,
498
.keep_alive = 60
499
};
500
+
501
#ifndef ACLK_DISABLE_CHALLENGE
477
- aclk_get_mqtt_otp(aclk_private_key, aclk_hostname, aclk_port, &mqtt_otp_user, &mqtt_otp_pass);
502
+ if (aclk_env) {
503
+ aclk_env_t_destroy(aclk_env);
504
+ freez(aclk_env);
505
+ }
506
+ aclk_env = callocz(1, sizeof(aclk_env_t));
507
+
508
+ ret = aclk_get_env(aclk_env, base_url.host, base_url.port);
509
+ url_t_destroy(&base_url);
510
+ if (ret) {
511
+ error("Failed to Get ACLK environment");
512
+ // delay handled by aclk_block_till_recon_allowed
513
+ continue;
514
+ }
515
+
516
+ memset(&auth_url, 0, sizeof(url_t));
517
+ if (url_parse(aclk_env->auth_endpoint, &auth_url)) {
518
+ error("Parsing URL returned by env endpoint for authentication failed. \"%s\"", aclk_env->auth_endpoint);
519
+ url_t_destroy(&auth_url);
520
+ continue;
521
+ }
522
+
523
+ // TODO check success
524
+ aclk_get_mqtt_otp(aclk_private_key, &mqtt_otp_user, &mqtt_otp_pass, &auth_url);
525
+ url_t_destroy(&auth_url);
526
+
527
mqtt_conn_params.clientid = mqtt_otp_user;
528
mqtt_conn_params.username = mqtt_otp_user;
529
mqtt_conn_params.password = mqtt_otp_pass;
530
+
531
+ // Do the MQTT connection
532
+ ret = aclk_get_transport_idx(aclk_env);
533
+ if (ret < 0) {
534
+ error("Cloud /env endpoint didn't return any transport usable by this Agent.");
535
+ continue;
536
+ }
537
+
538
+ memset(&mqtt_url, 0, sizeof(url_t));
539
+ if (url_parse(aclk_env->transports[ret]->endpoint, &mqtt_url)){
540
+ error("Failed to parse target URL for /env trp idx %d \"%s\"", ret, aclk_env->transports[ret]->endpoint);
541
+ url_t_destroy(&mqtt_url);
542
+ continue;
543
+ }
544
#endif
545
546
lwt = aclk_generate_disconnect(NULL);
547
mqtt_conn_params.will_msg = json_object_to_json_string_ext(lwt, JSON_C_TO_STRING_PLAIN);
485
-
548
mqtt_conn_params.will_msg_len = strlen(mqtt_conn_params.will_msg);
487
- if (!mqtt_wss_connect(client, aclk_hostname, aclk_port, &mqtt_conn_params, ACLK_SSL_FLAGS, &proxy_conf)) {
488
- json_object_put(lwt);
489
- freez(aclk_hostname);
490
- aclk_hostname = NULL;
549
+
550
+#ifdef ACLK_DISABLE_CHALLENGE
551
+ ret = mqtt_wss_connect(client, base_url.host, base_url.port, &mqtt_conn_params, ACLK_SSL_FLAGS, &proxy_conf);
552
+ url_t_destroy(&base_url);
553
+#else
554
+ ret = mqtt_wss_connect(client, mqtt_url.host, mqtt_url.port, &mqtt_conn_params, ACLK_SSL_FLAGS, &proxy_conf);
555
+ url_t_destroy(&mqtt_url);
556
+#endif
557
+
558
+ json_object_put(lwt);
559
+
560
+ if (!ret) {
561
info("MQTTWSS connection succeeded");
562
mqtt_connected_actions(client);
563
return 0;
564
}
565
496
- freez(aclk_hostname);
497
- aclk_hostname = NULL;
498
- json_object_put(lwt);
566
error("Connect failed\n");
567
}
568
@@ -597,6 +664,10 @@ exit_full:
664
free_topic_cache();
665
mqtt_wss_destroy(mqttwss_client);
666
exit:
667
+ if (aclk_env) {
668
+ aclk_env_t_destroy(aclk_env);
669
+ freez(aclk_env);
670
+ }
671
static_thread->enabled = NETDATA_MAIN_THREAD_EXITED;
672
return NULL;
673
}
aclk/aclk.h
+2
@@ -43,6 +43,8 @@ extern int aclk_connected;
43
extern usec_t aclk_session_us;
44
extern time_t aclk_session_sec;
45
46
+extern aclk_env_t *aclk_env;
47
+
48
void *aclk_main(void *ptr);
49
void aclk_single_update_disable();
50
void aclk_single_update_enable();
aclk/aclk_otp.c
+359
-7
@@ -3,8 +3,6 @@
3
4
#include "aclk_otp.h"
5
6
-#include "https_client.h"
7
-
6
#include "../daemon/common.h"
7
8
#include "../mqtt_websockets/c-rbuf/include/ringbuffer.h"
@@ -191,7 +189,7 @@ static int aclk_https_request(https_req_t *request, https_req_response_t *respon
189
}
190
191
#define OTP_URL_PREFIX "/api/v1/auth/node/"
194
-void aclk_get_mqtt_otp(RSA *p_key, char *aclk_hostname, int port, char **mqtt_usr, char **mqtt_pass) {
192
+void aclk_get_mqtt_otp(RSA *p_key, char **mqtt_usr, char **mqtt_pass, url_t *target) {
193
BUFFER *url = buffer_create(strlen(OTP_URL_PREFIX) + UUID_STR_LEN + 20);
194
195
https_req_t req = HTTPS_REQ_T_INITIALIZER;
@@ -205,9 +203,9 @@ void aclk_get_mqtt_otp(RSA *p_key, char *aclk_hostname, int port, char **mqtt_us
203
}
204
205
// GET Challenge
208
- req.host = aclk_hostname;
209
- req.port = port;
210
- buffer_sprintf(url, "%s%s/challenge", OTP_URL_PREFIX, agent_id);
206
+ req.host = target->host;
207
+ req.port = target->port;
208
+ buffer_sprintf(url, "%s/node/%s/challenge", target->path, agent_id);
209
req.url = url->buffer;
210
211
if (aclk_https_request(&req, &resp)) {
@@ -256,7 +254,7 @@ void aclk_get_mqtt_otp(RSA *p_key, char *aclk_hostname, int port, char **mqtt_us
254
// POST password
255
req.request_type = HTTP_REQ_POST;
256
buffer_flush(url);
259
- buffer_sprintf(url, "%s%s/password", OTP_URL_PREFIX, agent_id);
257
+ buffer_sprintf(url, "%s/node/%s/password", target->path, agent_id);
258
req.url = url->buffer;
259
req.payload = response_json;
260
req.payload_size = strlen(response_json);
@@ -298,3 +296,357 @@ cleanup:
296
freez(agent_id);
297
buffer_free(url);
298
}
299
+
300
+#define PARSE_ENV_JSON_CHK_TYPE(it, type, name) \
301
+ if (json_object_get_type(json_object_iter_peek_value(it)) != type) { \
302
+ error("value of key \"%s\" should be %s", name, #type); \
303
+ goto exit; \
304
+ }
305
+
306
+#define JSON_KEY_ENC "encoding"
307
+#define JSON_KEY_AUTH_ENDPOINT "authEndpoint"
308
+#define JSON_KEY_TRP "transports"
309
+#define JSON_KEY_TRP_TYPE "type"
310
+#define JSON_KEY_TRP_ENDPOINT "endpoint"
311
+#define JSON_KEY_BACKOFF "backoff"
312
+#define JSON_KEY_BACKOFF_BASE "base"
313
+#define JSON_KEY_BACKOFF_MAX "maxSeconds"
314
+#define JSON_KEY_BACKOFF_MIN "minSeconds"
315
+#define JSON_KEY_CAPS "capabilities"
316
+
317
+static int parse_json_env_transport(json_object *json, aclk_transport_desc_t *trp) {
318
+ struct json_object_iterator it;
319
+ struct json_object_iterator itEnd;
320
+
321
+ it = json_object_iter_begin(json);
322
+ itEnd = json_object_iter_end(json);
323
+
324
+ while (!json_object_iter_equal(&it, &itEnd)) {
325
+ if (!strcmp(json_object_iter_peek_name(&it), JSON_KEY_TRP_TYPE)) {
326
+ PARSE_ENV_JSON_CHK_TYPE(&it, json_type_string, JSON_KEY_TRP_TYPE)
327
+ if (trp->type != ACLK_TRP_UNKNOWN) {
328
+ error(JSON_KEY_TRP_TYPE " set already");
329
+ goto exit;
330
+ }
331
+ trp->type = aclk_transport_type_t_from_str(json_object_get_string(json_object_iter_peek_value(&it)));
332
+ if (trp->type == ACLK_TRP_UNKNOWN) {
333
+ error(JSON_KEY_TRP_TYPE " unknown type \"%s\"", json_object_get_string(json_object_iter_peek_value(&it)));
334
+ goto exit;
335
+ }
336
+ json_object_iter_next(&it);
337
+ continue;
338
+ }
339
+
340
+ if (!strcmp(json_object_iter_peek_name(&it), JSON_KEY_TRP_ENDPOINT)) {
341
+ PARSE_ENV_JSON_CHK_TYPE(&it, json_type_string, JSON_KEY_TRP_ENDPOINT)
342
+ if (trp->endpoint) {
343
+ error(JSON_KEY_TRP_ENDPOINT " set already");
344
+ goto exit;
345
+ }
346
+ trp->endpoint = strdupz(json_object_get_string(json_object_iter_peek_value(&it)));
347
+ json_object_iter_next(&it);
348
+ continue;
349
+ }
350
+
351
+ error ("unknown JSON key in dictionary (\"%s\")", json_object_iter_peek_name(&it));
352
+ json_object_iter_next(&it);
353
+ }
354
+
355
+ if (!trp->endpoint) {
356
+ error (JSON_KEY_TRP_ENDPOINT " is missing from JSON dictionary");
357
+ goto exit;
358
+ }
359
+
360
+ if (trp->type == ACLK_TRP_UNKNOWN) {
361
+ error ("transport type not set");
362
+ goto exit;
363
+ }
364
+
365
+ return 0;
366
+
367
+exit:
368
+ aclk_transport_desc_t_destroy(trp);
369
+ return 1;
370
+}
371
+
372
+static int parse_json_env_transports(json_object *json_array, aclk_env_t *env) {
373
+ aclk_transport_desc_t *trp;
374
+ json_object *obj;
375
+
376
+ if (env->transports) {
377
+ error("transports have been set already");
378
+ return 1;
379
+ }
380
+
381
+ env->transport_count = json_object_array_length(json_array);
382
+
383
+ env->transports = callocz(env->transport_count , sizeof(aclk_transport_desc_t *));
384
+
385
+ for (size_t i = 0; i < env->transport_count; i++) {
386
+ trp = callocz(1, sizeof(aclk_transport_desc_t));
387
+ obj = json_object_array_get_idx(json_array, i);
388
+ if (parse_json_env_transport(obj, trp)) {
389
+ error("error parsing transport idx %d", (int)i);
390
+ freez(trp);
391
+ return 1;
392
+ }
393
+ env->transports[i] = trp;
394
+ }
395
+
396
+ return 0;
397
+}
398
+
399
+#define MATCHED_CORRECT 1
400
+#define MATCHED_ERROR -1
401
+#define NOT_MATCHED 0
402
+static int parse_json_backoff_int(struct json_object_iterator *it, int *out, const char* name, int min, int max) {
403
+ if (!strcmp(json_object_iter_peek_name(it), name)) {
404
+ if (json_object_get_type(json_object_iter_peek_value(it)) != json_type_int) {
405
+ error("Could not parse \"%s\". Not an integer as expected.", name);
406
+ return MATCHED_ERROR;
407
+ }
408
+
409
+ *out = json_object_get_int(json_object_iter_peek_value(it));
410
+
411
+ if (*out < min || *out > max) {
412
+ error("Value of \"%s\"=%d out of range (%d-%d).", name, *out, min, max);
413
+ return MATCHED_ERROR;
414
+ }
415
+
416
+ return MATCHED_CORRECT;
417
+ }
418
+ return NOT_MATCHED;
419
+}
420
+
421
+static int parse_json_backoff(json_object *json, aclk_backoff_t *backoff) {
422
+ struct json_object_iterator it;
423
+ struct json_object_iterator itEnd;
424
+ int ret;
425
+
426
+ it = json_object_iter_begin(json);
427
+ itEnd = json_object_iter_end(json);
428
+
429
+ while (!json_object_iter_equal(&it, &itEnd)) {
430
+ if ( (ret = parse_json_backoff_int(&it, &backoff->base, JSON_KEY_BACKOFF_BASE, 1, 10)) ) {
431
+ if (ret == MATCHED_ERROR) {
432
+ return 1;
433
+ }
434
+ json_object_iter_next(&it);
435
+ continue;
436
+ }
437
+
438
+ if ( (ret = parse_json_backoff_int(&it, &backoff->max_s, JSON_KEY_BACKOFF_MAX, 500, INT_MAX)) ) {
439
+ if (ret == MATCHED_ERROR) {
440
+ return 1;
441
+ }
442
+ json_object_iter_next(&it);
443
+ continue;
444
+ }
445
+
446
+ if ( (ret = parse_json_backoff_int(&it, &backoff->min_s, JSON_KEY_BACKOFF_MIN, 0, INT_MAX)) ) {
447
+ if (ret == MATCHED_ERROR) {
448
+ return 1;
449
+ }
450
+ json_object_iter_next(&it);
451
+ continue;
452
+ }
453
+
454
+ error ("unknown JSON key in dictionary (\"%s\")", json_object_iter_peek_name(&it));
455
+ json_object_iter_next(&it);
456
+ }
457
+
458
+ return 0;
459
+}
460
+
461
+static int parse_json_env_caps(json_object *json, aclk_env_t *env) {
462
+ json_object *obj;
463
+ const char *str;
464
+
465
+ if (env->capabilities) {
466
+ error("transports have been set already");
467
+ return 1;
468
+ }
469
+
470
+ env->capability_count = json_object_array_length(json);
471
+
472
+ // empty capabilities list is allowed
473
+ if (!env->capability_count)
474
+ return 0;
475
+
476
+ env->capabilities = callocz(env->capability_count , sizeof(char *));
477
+
478
+ for (size_t i = 0; i < env->capability_count; i++) {
479
+ obj = json_object_array_get_idx(json, i);
480
+ if (json_object_get_type(obj) != json_type_string) {
481
+ error("Capability at index %d not a string!", (int)i);
482
+ return 1;
483
+ }
484
+ str = json_object_get_string(obj);
485
+ if (!str) {
486
+ error("Error parsing capabilities");
487
+ return 1;
488
+ }
489
+ env->capabilities[i] = strdupz(str);
490
+ }
491
+
492
+ return 0;
493
+}
494
+
495
+static int parse_json_env(const char *json_str, aclk_env_t *env) {
496
+ json_object *json;
497
+ struct json_object_iterator it;
498
+ struct json_object_iterator itEnd;
499
+
500
+ json = json_tokener_parse(json_str);
501
+ if (!json) {
502
+ error("JSON-C failed to parse the payload of http respons of /env endpoint");
503
+ return 1;
504
+ }
505
+
506
+ it = json_object_iter_begin(json);
507
+ itEnd = json_object_iter_end(json);
508
+
509
+ while (!json_object_iter_equal(&it, &itEnd)) {
510
+ if (!strcmp(json_object_iter_peek_name(&it), JSON_KEY_AUTH_ENDPOINT)) {
511
+ PARSE_ENV_JSON_CHK_TYPE(&it, json_type_string, JSON_KEY_AUTH_ENDPOINT)
512
+ if (env->auth_endpoint) {
513
+ error("authEndpoint set already");
514
+ goto exit;
515
+ }
516
+ env->auth_endpoint = strdupz(json_object_get_string(json_object_iter_peek_value(&it)));
517
+ json_object_iter_next(&it);
518
+ continue;
519
+ }
520
+
521
+ if (!strcmp(json_object_iter_peek_name(&it), JSON_KEY_ENC)) {
522
+ PARSE_ENV_JSON_CHK_TYPE(&it, json_type_string, JSON_KEY_ENC)
523
+ if (env->encoding != ACLK_ENC_UNKNOWN) {
524
+ error(JSON_KEY_ENC " set already");
525
+ goto exit;
526
+ }
527
+ env->encoding = aclk_encoding_type_t_from_str(json_object_get_string(json_object_iter_peek_value(&it)));
528
+ json_object_iter_next(&it);
529
+ continue;
530
+ }
531
+
532
+ if (!strcmp(json_object_iter_peek_name(&it), JSON_KEY_TRP)) {
533
+ PARSE_ENV_JSON_CHK_TYPE(&it, json_type_array, JSON_KEY_TRP)
534
+
535
+ json_object *now = json_object_iter_peek_value(&it);
536
+ parse_json_env_transports(now, env);
537
+
538
+ json_object_iter_next(&it);
539
+ continue;
540
+ }
541
+
542
+ if (!strcmp(json_object_iter_peek_name(&it), JSON_KEY_BACKOFF)) {
543
+ PARSE_ENV_JSON_CHK_TYPE(&it, json_type_object, JSON_KEY_BACKOFF)
544
+
545
+ if (parse_json_backoff(json_object_iter_peek_value(&it), &env->backoff)) {
546
+ env->backoff.base = 0;
547
+ error("Error parsing Backoff parameters in env");
548
+ goto exit;
549
+ }
550
+
551
+ json_object_iter_next(&it);
552
+ continue;
553
+ }
554
+
555
+ if (!strcmp(json_object_iter_peek_name(&it), JSON_KEY_CAPS)) {
556
+ PARSE_ENV_JSON_CHK_TYPE(&it, json_type_array, JSON_KEY_CAPS)
557
+
558
+ if (parse_json_env_caps(json_object_iter_peek_value(&it), env)) {
559
+ error("Error parsing capabilities list");
560
+ goto exit;
561
+ }
562
+
563
+ json_object_iter_next(&it);
564
+ continue;
565
+ }
566
+
567
+ error ("unknown JSON key in dictionary (\"%s\")", json_object_iter_peek_name(&it));
568
+ json_object_iter_next(&it);
569
+ }
570
+
571
+ // Check all compulsory keys have been set
572
+ if (env->transport_count < 1) {
573
+ error("env has to return at least one transport");
574
+ goto exit;
575
+ }
576
+ if (!env->auth_endpoint) {
577
+ error(JSON_KEY_AUTH_ENDPOINT " is compulsory");
578
+ goto exit;
579
+ }
580
+ if (env->encoding == ACLK_ENC_UNKNOWN) {
581
+ error(JSON_KEY_ENC " is compulsory");
582
+ goto exit;
583
+ }
584
+ if (!env->backoff.base) {
585
+ error(JSON_KEY_BACKOFF " is compulsory");
586
+ goto exit;
587
+ }
588
+
589
+ json_object_put(json);
590
+ return 0;
591
+
592
+exit:
593
+ aclk_env_t_destroy(env);
594
+ json_object_put(json);
595
+ return 1;
596
+}
597
+
598
+int aclk_get_env(aclk_env_t *env, const char* aclk_hostname, int aclk_port) {
599
+ BUFFER *buf = buffer_create(1024);
600
+
601
+ https_req_t req = HTTPS_REQ_T_INITIALIZER;
602
+ https_req_response_t resp = HTTPS_REQ_RESPONSE_T_INITIALIZER;
603
+
604
+ req.request_type = HTTP_REQ_GET;
605
+
606
+ char *agent_id = is_agent_claimed();
607
+ if (agent_id == NULL)
608
+ {
609
+ error("Agent was not claimed - cannot perform challenge/response");
610
+ buffer_free(buf);
611
+ return 1;
612
+ }
613
+
614
+ buffer_sprintf(buf, "/api/v1/env?v=%s&cap=json$claim_id=%s", &(VERSION[1]) /* skip 'v' at beginning */, agent_id);
615
+ freez(agent_id);
616
+
617
+ req.host = (char*)aclk_hostname;
618
+ req.port = aclk_port;
619
+ req.url = buf->buffer;
620
+ if (aclk_https_request(&req, &resp)) {
621
+ error("Error trying to contact env endpoint");
622
+ https_req_response_free(&resp);
623
+ buffer_free(buf);
624
+ return 1;
625
+ }
626
+ if (resp.http_code != 200) {
627
+ error("The HTTP code not 200 OK (Got %d)", resp.http_code);
628
+ https_req_response_free(&resp);
629
+ buffer_free(buf);
630
+ return 1;
631
+ }
632
+
633
+ if (!resp.payload || !resp.payload_size) {
634
+ error("Unexpected empty payload as response to /env call");
635
+ https_req_response_free(&resp);
636
+ buffer_free(buf);
637
+ return 1;
638
+ }
639
+
640
+ if (parse_json_env(resp.payload, env)) {
641
+ error ("error parsing /env message");
642
+ https_req_response_free(&resp);
643
+ buffer_free(buf);
644
+ return 1;
645
+ }
646
+
647
+ info("Getting Cloud /env successful");
648
+
649
+ https_req_response_free(&resp);
650
+ buffer_free(buf);
651
+ return 0;
652
+}
aclk/aclk_otp.h
+4
-1
@@ -5,6 +5,9 @@
5
6
#include "../daemon/common.h"
7
8
-void aclk_get_mqtt_otp(RSA *p_key, char *aclk_hostname, int port, char **mqtt_usr, char **mqtt_pass);
8
+#include "https_client.h"
9
+
10
+void aclk_get_mqtt_otp(RSA *p_key, char **mqtt_usr, char **mqtt_pass, url_t *target);
11
+int aclk_get_env(aclk_env_t *env, const char *aclk_hostname, int aclk_port);
12
13
#endif /* ACLK_OTP_H */
aclk/aclk_util.c
+42
-41
@@ -10,6 +10,48 @@
10
#define UUID_STR_LEN 37
11
#endif
12
13
+aclk_encoding_type_t aclk_encoding_type_t_from_str(const char *str) {
14
+ if (!strcmp(str, "json")) {
15
+ return ACLK_ENC_JSON;
16
+ }
17
+ if (!strcmp(str, "proto")) {
18
+ return ACLK_ENC_PROTO;
19
+ }
20
+ return ACLK_ENC_UNKNOWN;
21
+}
22
+
23
+aclk_transport_type_t aclk_transport_type_t_from_str(const char *str) {
24
+ if (!strcmp(str, "MQTTv3")) {
25
+ return ACLK_TRP_MQTT_3_1_1;
26
+ }
27
+ if (!strcmp(str, "MQTTv5")) {
28
+ return ACLK_TRP_MQTT_5;
29
+ }
30
+ return ACLK_TRP_UNKNOWN;
31
+}
32
+
33
+void aclk_transport_desc_t_destroy(aclk_transport_desc_t *trp_desc) {
34
+ freez(trp_desc->endpoint);
35
+}
36
+
37
+void aclk_env_t_destroy(aclk_env_t *env) {
38
+ freez(env->auth_endpoint);
39
+ if (env->transports) {
40
+ for (size_t i = 0; i < env->transport_count; i++) {
41
+ if(env->transports[i]) {
42
+ aclk_transport_desc_t_destroy(env->transports[i]);
43
+ env->transports[i] = NULL;
44
+ }
45
+ }
46
+ freez(env->transports);
47
+ }
48
+ if (env->capabilities) {
49
+ for (size_t i = 0; i < env->capability_count; i++)
50
+ freez(env->capabilities[i]);
51
+ freez(env->capabilities);
52
+ }
53
+}
54
+
55
#ifdef ACLK_LOG_CONVERSATION_DIR
56
volatile int aclk_conversation_log_counter = 0;
57
#if !defined(HAVE_C___ATOMIC) || defined(NETDATA_NO_ATOMIC_INSTRUCTIONS)
@@ -88,47 +130,6 @@ const char *aclk_get_topic(enum aclk_topics topic)
130
return aclk_topic_cache[topic].topic;
131
}
132
91
-int aclk_decode_base_url(char *url, char **aclk_hostname, int *aclk_port)
92
-{
93
- int pos = 0;
94
- if (!strncmp("https://", url, 8)) {
95
- pos = 8;
96
- } else if (!strncmp("http://", url, 7)) {
97
- error("Cannot connect ACLK over %s -> unencrypted link is not supported", url);
98
- return 1;
99
- }
100
- int host_end = pos;
101
- while (url[host_end] != 0 && url[host_end] != '/' && url[host_end] != ':')
102
- host_end++;
103
- if (url[host_end] == 0) {
104
- *aclk_hostname = strdupz(url + pos);
105
- *aclk_port = 443;
106
- info("Setting ACLK target host=%s port=%d from %s", *aclk_hostname, *aclk_port, url);
107
- return 0;
108
- }
109
- if (url[host_end] == ':') {
110
- *aclk_hostname = callocz(host_end - pos + 1, 1);
111
- strncpy(*aclk_hostname, url + pos, host_end - pos);
112
- int port_end = host_end + 1;
113
- while (url[port_end] >= '0' && url[port_end] <= '9')
114
- port_end++;
115
- if (port_end - host_end > 6) {
116
- error("Port specified in %s is invalid", url);
117
- freez(*aclk_hostname);
118
- *aclk_hostname = NULL;
119
- return 1;
120
- }
121
- *aclk_port = atoi(&url[host_end+1]);
122
- }
123
- if (url[host_end] == '/') {
124
- *aclk_port = 443;
125
- *aclk_hostname = callocz(1, host_end - pos + 1);
126
- strncpy(*aclk_hostname, url+pos, host_end - pos);
127
- }
128
- info("Setting ACLK target host=%s port=%d from %s", *aclk_hostname, *aclk_port, url);
129
- return 0;
130
-}
131
-
133
/*
134
* TBEB with randomness
135
*
aclk/aclk_util.h
+41
-2
@@ -8,7 +8,47 @@
8
// Helper stuff which should not have any further inside ACLK dependency
9
// and are supposed not to be needed outside of ACLK
10
11
-int aclk_decode_base_url(char *url, char **aclk_hostname, int *aclk_port);
11
+typedef enum {
12
+ ACLK_ENC_UNKNOWN = 0,
13
+ ACLK_ENC_JSON,
14
+ ACLK_ENC_PROTO
15
+} aclk_encoding_type_t;
16
+
17
+typedef enum {
18
+ ACLK_TRP_UNKNOWN = 0,
19
+ ACLK_TRP_MQTT_3_1_1,
20
+ ACLK_TRP_MQTT_5
21
+} aclk_transport_type_t;
22
+
23
+typedef struct {
24
+ char *endpoint;
25
+ aclk_transport_type_t type;
26
+} aclk_transport_desc_t;
27
+
28
+typedef struct {
29
+ int base;
30
+ int max_s;
31
+ int min_s;
32
+} aclk_backoff_t;
33
+
34
+typedef struct {
35
+ char *auth_endpoint;
36
+ aclk_encoding_type_t encoding;
37
+
38
+ aclk_transport_desc_t **transports;
39
+ size_t transport_count;
40
+
41
+ char **capabilities;
42
+ size_t capability_count;
43
+
44
+ aclk_backoff_t backoff;
45
+} aclk_env_t;
46
+
47
+aclk_encoding_type_t aclk_encoding_type_t_from_str(const char *str);
48
+aclk_transport_type_t aclk_transport_type_t_from_str(const char *str);
49
+
50
+void aclk_transport_desc_t_destroy(aclk_transport_desc_t *trp_desc);
51
+void aclk_env_t_destroy(aclk_env_t *env);
52
53
enum aclk_topics {
54
ACLK_TOPICID_CHART = 0,
@@ -47,7 +87,6 @@ const char *aclk_proxy_type_to_s(ACLK_PROXY_TYPE *type);
87
ACLK_PROXY_TYPE aclk_verify_proxy(const char *string);
88
const char *aclk_lws_wss_get_proxy_setting(ACLK_PROXY_TYPE *type);
89
void safe_log_proxy_censor(char *proxy);
50
-int aclk_decode_base_url(char *url, char **aclk_hostname, int *aclk_port);
90
const char *aclk_get_proxy(ACLK_PROXY_TYPE *type);
91
92
void aclk_set_proxy(char **ohost, int *port, enum mqtt_wss_proxy_type *type);
aclk/https_client.c
+100
@@ -550,3 +550,103 @@ void https_req_response_init(https_req_response_t *res) {
550
res->payload = NULL;
551
res->payload_size = 0;
552
}
553
+
554
+static inline char *min_non_null(char *a, char *b) {
555
+ if (!a)
556
+ return b;
557
+ if (!b)
558
+ return a;
559
+ return (a < b ? a : b);
560
+}
561
+
562
+#define URI_PROTO_SEPARATOR "://"
563
+#define URL_PARSER_LOG_PREFIX "url_parser "
564
+
565
+static int parse_host_port(url_t *url) {
566
+ char *ptr = strrchr(url->host, ':');
567
+ if (ptr) {
568
+ size_t port_len = strlen(ptr + 1);
569
+ if (!port_len) {
570
+ error(URL_PARSER_LOG_PREFIX ": specified but no port number");
571
+ return 1;
572
+ }
573
+ if (port_len > 5 /* MAX port lenght is 5digit long in decimal */) {
574
+ error(URL_PARSER_LOG_PREFIX "port # is too long");
575
+ return 1;
576
+ }
577
+ *ptr = 0;
578
+ if (!strlen(url->host)) {
579
+ error(URL_PARSER_LOG_PREFIX "host empty after removing port");
580
+ return 1;
581
+ }
582
+ url->port = atoi (ptr + 1);
583
+ }
584
+ return 0;
585
+}
586
+
587
+static inline void port_by_proto(url_t *url) {
588
+ if (url->port)
589
+ return;
590
+ if (!url->proto)
591
+ return;
592
+ if (!strcmp(url->proto, "http")) {
593
+ url->port = 80;
594
+ return;
595
+ }
596
+ if (!strcmp(url->proto, "https")) {
597
+ url->port = 443;
598
+ return;
599
+ }
600
+}
601
+
602
+#define STRDUPZ_2PTR(dest, start, end) \
603
+ { \
604
+ dest = mallocz(1 + end - start); \
605
+ memcpy(dest, start, end - start); \
606
+ dest[end - start] = 0; \
607
+ }
608
+
609
+int url_parse(const char *url, url_t *parsed) {
610
+ const char *start = url;
611
+ const char *end = strstr(url, URI_PROTO_SEPARATOR);
612
+
613
+ if (end) {
614
+ if (end == start) {
615
+ error (URL_PARSER_LOG_PREFIX "found " URI_PROTO_SEPARATOR " without protocol specified");
616
+ return 1;
617
+ }
618
+
619
+ STRDUPZ_2PTR(parsed->proto, start, end)
620
+ start = end + strlen(URI_PROTO_SEPARATOR);
621
+ }
622
+
623
+ end = strchr(start, '/');
624
+ if (!end)
625
+ end = start + strlen(start);
626
+
627
+ if (start == end) {
628
+ error(URL_PARSER_LOG_PREFIX "Host empty");
629
+ return 1;
630
+ }
631
+
632
+ STRDUPZ_2PTR(parsed->host, start, end);
633
+
634
+ if (parse_host_port(parsed))
635
+ return 1;
636
+
637
+ if (!*end) {
638
+ parsed->path = strdupz("/");
639
+ port_by_proto(parsed);
640
+ return 0;
641
+ }
642
+
643
+ parsed->path = strdupz(end);
644
+ port_by_proto(parsed);
645
+ return 0;
646
+}
647
+
648
+void url_t_destroy(url_t *url) {
649
+ freez(url->host);
650
+ freez(url->path);
651
+ freez(url->proto);
652
+}
aclk/https_client.h
+16
@@ -34,6 +34,22 @@ typedef struct {
34
size_t payload_size;
35
} https_req_response_t;
36
37
+
38
+// Non feature complete URL parser
39
+// feel free to extend when needed
40
+// currently implements only what ACLK
41
+// needs
42
+// proto://host[:port]/path
43
+typedef struct {
44
+ char *proto;
45
+ char *host;
46
+ int port;
47
+ char* path;
48
+} url_t;
49
+
50
+int url_parse(const char *url, url_t *parsed);
51
+void url_t_destroy(url_t *url);
52
+
53
void https_req_response_free(https_req_response_t *res);
54
void https_req_response_init(https_req_response_t *res);
55