Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

Add support for SSL registry #41

Open
wants to merge 1 commit into
base: master
Choose a base branch
from
Open

Add support for SSL registry #41

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 4 additions & 0 deletions README.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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`)
28 changes: 24 additions & 4 deletions src/rest.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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) {

Expand Down Expand Up @@ -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;
Expand Down Expand Up @@ -462,26 +479,29 @@ 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;
}


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;
Expand Down
16 changes: 15 additions & 1 deletion src/rest.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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'.
Expand Down Expand Up @@ -101,14 +112,17 @@ 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.
*
* 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, ...);

Expand Down
6 changes: 5 additions & 1 deletion src/schema-cache.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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);
Expand Down Expand Up @@ -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);
}

Expand Down
52 changes: 52 additions & 0 deletions src/serdes.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,7 @@
#include "serdes_int.h"

#include <stdarg.h>
#include <curl/curl.h>

const char *serdes_err2str (serdes_err_t err) {
switch (err)
Expand Down Expand Up @@ -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) {
Expand All @@ -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) {
Expand Down Expand Up @@ -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;
Expand Down
1 change: 1 addition & 0 deletions src/serdes_int.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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 */


Expand Down