diff --git a/README.md b/README.md index b13d125..0078165 100644 --- a/README.md +++ b/README.md @@ -66,4 +66,8 @@ to the schema registries, all other configuration is optional. * `schema.registry.url` - comma separated list of schema registry base URLs (no default) * `deserializer.framing` - expected framing format when deserializing data: `none` or `cp1` (Confluent Platform framing). (default: `cp1`) * `serializer.framing` - framing format inserted when serializing data: `none` or `cp1` (Confluent Platform framing). (default: `cp1`) + * `ssl.ca.location` - location to the CA certificate (no default) + * `ssl.certificate.location` - location to the client certificate. (no default) + * `ssl.key.location` - location to the private key. (no default) + * `ssl.enabled.min_protocol` - the minimal tls version to use. The value can be `1.0`, `1.1`, `1.2` or `1.3`. (no default) * `debug` - enable/disable debugging with `all` or `none`. (default: `none`) diff --git a/src/rest.c b/src/rest.c index bedde11..afe05cf 100644 --- a/src/rest.c +++ b/src/rest.c @@ -358,7 +358,9 @@ static CURLcode rest_req_curl (CURL *curl, rest_response_t *rr) { * * Returns a response handle which needs to be checked for error. */ -static rest_response_t *rest_req (url_list_t *ul, rest_cmd_t cmd, +static rest_response_t *rest_req (url_list_t *ul, + const security_info_t *secure_info, + rest_cmd_t cmd, const void *payload, int size, const char *url_path_fmt, va_list ap) { @@ -427,6 +429,21 @@ static rest_response_t *rest_req (url_list_t *ul, rest_cmd_t cmd, break; } + if (secure_info) { + if (secure_info->ca_path && strlen(secure_info->ca_path)) { + do_curl_setopt(curl, CURLOPT_CAINFO, secure_info->ca_path); + } + if (secure_info->cert_path && strlen(secure_info->cert_path)) { + do_curl_setopt(curl, CURLOPT_SSLCERT, secure_info->cert_path); + } + if (secure_info->key_path && strlen(secure_info->key_path)) { + do_curl_setopt(curl, CURLOPT_SSLKEY, secure_info->key_path); + } + if (secure_info->min_tls_version) { + do_curl_setopt(curl, CURLOPT_SSLVERSION, secure_info->min_tls_version); + } + } + /* Try each URL in the URL list until one works. */ ccode = CURLE_URL_MALFORMAT; @@ -462,12 +479,14 @@ static rest_response_t *rest_req (url_list_t *ul, rest_cmd_t cmd, -rest_response_t *rest_get (url_list_t *ul, const char *url_path_fmt, ...) { +rest_response_t *rest_get (url_list_t *ul, + const security_info_t *secure_info, + const char *url_path_fmt, ...) { rest_response_t *rr; va_list ap; va_start(ap, url_path_fmt); - rr = rest_req(ul, REST_GET, NULL, 0, url_path_fmt, ap); + rr = rest_req(ul, secure_info, REST_GET, NULL, 0, url_path_fmt, ap); va_end(ap); return rr; @@ -475,13 +494,14 @@ rest_response_t *rest_get (url_list_t *ul, const char *url_path_fmt, ...) { rest_response_t *rest_post (url_list_t *ul, + const security_info_t *secure_info, const void *payload, int size, const char *url_path_fmt, ...) { rest_response_t *rr; va_list ap; va_start(ap, url_path_fmt); - rr = rest_req(ul, REST_POST, payload, size, url_path_fmt, ap); + rr = rest_req(ul, secure_info, REST_POST, payload, size, url_path_fmt, ap); va_end(ap); return rr; diff --git a/src/rest.h b/src/rest.h index 14149b2..e4568a4 100644 --- a/src/rest.h +++ b/src/rest.h @@ -36,6 +36,17 @@ typedef struct url_list_s { int max_len; /* Longest URL's length */ } url_list_t; +/** + * HTTP security related information. SSL/Auth/etc. + */ +typedef struct security_info_s { + char *ca_path; + char *cert_path; + char *key_path; + /* CURLOPT_SSLVERSION */ + int min_tls_version; +} security_info_t; + /** * Parse a comma-separated list of URLs and store them in the provided 'ul'. @@ -101,7 +112,9 @@ void rest_response_destroy (rest_response_t *rr); * * This is a blocking call. */ -rest_response_t *rest_get (url_list_t *ul, const char *url_path_fmt, ...); +rest_response_t *rest_get (url_list_t *ul, + const security_info_t *secure_info, + const char *url_path_fmt, ...); /* REST PUT request. @@ -109,6 +122,7 @@ rest_response_t *rest_get (url_list_t *ul, const char *url_path_fmt, ...); * Same semantics as `rest_get()` but POSTs `payload` of `size` bytes. */ rest_response_t *rest_post (url_list_t *ul, + const security_info_t *secure_info, const void *payload, int size, const char *url_path_fmt, ...); diff --git a/src/schema-cache.c b/src/schema-cache.c index 06b611a..3db6592 100644 --- a/src/schema-cache.c +++ b/src/schema-cache.c @@ -120,7 +120,9 @@ static int serdes_schema_store (serdes_schema_t *ss, enc_len = strlen(enc); /* POST schema definition to remote schema registry */ - rr = rest_post(&sd->sd_conf.schema_registry_urls, enc, enc_len, + rr = rest_post(&sd->sd_conf.schema_registry_urls, + &sd->sd_conf.schema_security_info, + enc, enc_len, "/subjects/%s/versions", ss->ss_name); free(enc); @@ -241,10 +243,12 @@ static int serdes_schema_fetch (serdes_schema_t *ss, if (ss->ss_id != -1) { /* GET schema definition by id from remote schema registry */ rr = rest_get(&sd->sd_conf.schema_registry_urls, + &sd->sd_conf.schema_security_info, "/schemas/ids/%d", ss->ss_id); } else { /* GET schema definition by name from remote schema registry */ rr = rest_get(&sd->sd_conf.schema_registry_urls, + &sd->sd_conf.schema_security_info, "/subjects/%s/versions/latest", ss->ss_name); } diff --git a/src/serdes.c b/src/serdes.c index e730a7e..1c2a2ae 100644 --- a/src/serdes.c +++ b/src/serdes.c @@ -17,6 +17,7 @@ #include "serdes_int.h" #include +#include const char *serdes_err2str (serdes_err_t err) { switch (err) @@ -49,6 +50,12 @@ const char *serdes_err2str (serdes_err_t err) { static void serdes_conf_destroy0 (serdes_conf_t *sconf) { url_list_clear(&sconf->schema_registry_urls); + if (sconf->schema_security_info.ca_path) + free(sconf->schema_security_info.ca_path); + if (sconf->schema_security_info.cert_path) + free(sconf->schema_security_info.cert_path); + if (sconf->schema_security_info.key_path) + free(sconf->schema_security_info.key_path); } void serdes_conf_destroy (serdes_conf_t *sconf) { @@ -73,6 +80,24 @@ static void serdes_conf_copy0 (serdes_conf_t *dst, const serdes_conf_t *src) { dst->schema_unload_cb = src->schema_unload_cb; dst->log_cb = src->log_cb; dst->opaque = src->opaque; + + security_info_t *dst_info = &dst->schema_security_info; + const security_info_t *src_info = &src->schema_security_info; + if (dst_info->ca_path) { + free(dst_info->ca_path); + } + dst_info->ca_path = src_info->ca_path ? + strdup(src_info->ca_path) : NULL; + if (dst_info->cert_path) { + free(dst_info->cert_path); + } + dst_info->cert_path = src_info->cert_path ? + strdup(src_info->cert_path) : NULL; + if (dst_info->key_path) { + free(dst_info->key_path); + } + dst_info->key_path = src_info->key_path ? + strdup(src_info->key_path) : NULL; } serdes_conf_t *serdes_conf_copy (const serdes_conf_t *src) { @@ -114,6 +139,33 @@ serdes_err_t serdes_conf_set (serdes_conf_t *sconf, else sconf->deserializer_framing = framing; + } else if (!strcmp(name, "ssl.ca.location")) { + if (sconf->schema_security_info.ca_path) + free(sconf->schema_security_info.ca_path); + sconf->schema_security_info.ca_path = strdup(val); + } else if (!strcmp(name, "ssl.certificate.location") && val) { + if (sconf->schema_security_info.cert_path) + free(sconf->schema_security_info.cert_path); + sconf->schema_security_info.cert_path = strdup(val); + } else if (!strcmp(name, "ssl.key.location")) { + if (sconf->schema_security_info.key_path) + free(sconf->schema_security_info.key_path); + sconf->schema_security_info.key_path = strdup(val); + } else if (!strcmp(name, "ssl.enabled.min_protocol")) { + if (!strcmp(val, "1.0")) { + sconf->schema_security_info.min_tls_version = CURL_SSLVERSION_TLSv1_0; + } else if (!strcmp(val, "1.1")) { + sconf->schema_security_info.min_tls_version = CURL_SSLVERSION_TLSv1_1; + } else if (!strcmp(val, "1.2")) { + sconf->schema_security_info.min_tls_version = CURL_SSLVERSION_TLSv1_2; + } else if (!strcmp(val, "1.3")) { + sconf->schema_security_info.min_tls_version = CURL_SSLVERSION_TLSv1_3; + } else { + snprintf(errstr, errstr_size, + "Invalid value for %s, allowed values: " + "1.0, 1.1, 1.2, 1.3", name); + return SERDES_ERR_CONF_INVALID; + } } else if (!strcmp(name, "debug")) { if (!strcmp(val, "all")) sconf->debug = 1; diff --git a/src/serdes_int.h b/src/serdes_int.h index d2f4e97..e6bd9d0 100644 --- a/src/serdes_int.h +++ b/src/serdes_int.h @@ -55,6 +55,7 @@ typedef enum { struct serdes_conf_s { url_list_t schema_registry_urls; /* CSV list of schema * registry URLs. */ + security_info_t schema_security_info; /* Security information (SSL/Auth/etc.) */ int debug; /* Debugging 1=enabled */