| /* |
| * Licensed to the Apache Software Foundation (ASF) under one |
| * or more contributor license agreements. See the NOTICE file |
| * distributed with this work for additional information |
| * regarding copyright ownership. The ASF licenses this file |
| * to you under the Apache License, Version 2.0 (the |
| * "License"); you may not use this file except in compliance |
| * with the License. You may obtain a copy of the License at |
| * |
| * http://www.apache.org/licenses/LICENSE-2.0 |
| * |
| * Unless required by applicable law or agreed to in writing, |
| * software distributed under the License is distributed on an |
| * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY |
| * KIND, either express or implied. See the License for the |
| * specific language governing permissions and limitations |
| * under the License. |
| */ |
| |
| #include "qpid/dispatch/connection_manager.h" |
| |
| #include "connection_manager_private.h" |
| #include "dispatch_private.h" |
| #include "entity.h" |
| #include "server_private.h" |
| |
| #include "qpid/dispatch/ctools.h" |
| #include "qpid/dispatch/failoverlist.h" |
| #include "qpid/dispatch/threading.h" |
| |
| #include <proton/listener.h> |
| |
| #include <errno.h> |
| #include <inttypes.h> |
| #include <stdio.h> |
| #include <string.h> |
| |
| struct qd_config_ssl_profile_t { |
| DEQ_LINKS(qd_config_ssl_profile_t); |
| char *name; |
| char *ssl_password; |
| char *ssl_trusted_certificate_db; |
| char *ssl_uid_format; |
| char *uid_name_mapping_file; |
| char *ssl_certificate_file; |
| char *ssl_private_key_file; |
| char *ssl_ciphers; |
| char *ssl_protocols; |
| }; |
| |
| DEQ_DECLARE(qd_config_ssl_profile_t, qd_config_ssl_profile_list_t); |
| |
| struct qd_config_sasl_plugin_t { |
| DEQ_LINKS(qd_config_sasl_plugin_t); |
| char *name; |
| char *auth_service; |
| char *hostname; |
| char *sasl_init_hostname; |
| char *auth_ssl_profile; |
| }; |
| |
| DEQ_DECLARE(qd_config_sasl_plugin_t, qd_config_sasl_plugin_list_t); |
| |
| struct qd_connection_manager_t { |
| qd_log_source_t *log_source; |
| qd_server_t *server; |
| qd_listener_list_t listeners; |
| qd_connector_list_t connectors; |
| qd_config_ssl_profile_list_t config_ssl_profiles; |
| qd_config_sasl_plugin_list_t config_sasl_plugins; |
| }; |
| |
| const char *qd_log_message_components[] = |
| {"message-id", |
| "user-id", |
| "to", |
| "subject", |
| "reply-to", |
| "correlation-id", |
| "content-type", |
| "content-encoding", |
| "absolute-expiry-time", |
| "creation-time", |
| "group-id", |
| "group-sequence", |
| "reply-to-group-id", |
| "app-properties", |
| 0}; |
| |
| const char *ALL = "all"; |
| const char *NONE = "none"; |
| |
| /** |
| * Search the list of config_ssl_profiles for an ssl-profile that matches the passed in name |
| */ |
| static qd_config_ssl_profile_t *qd_find_ssl_profile(qd_connection_manager_t *cm, char *name) |
| { |
| qd_config_ssl_profile_t *ssl_profile = DEQ_HEAD(cm->config_ssl_profiles); |
| while (ssl_profile) { |
| if (strcmp(ssl_profile->name, name) == 0) |
| return ssl_profile; |
| ssl_profile = DEQ_NEXT(ssl_profile); |
| } |
| |
| return 0; |
| } |
| |
| /** |
| * Read the file from the password_file location on the file system and populate password_field with the |
| * contents of the file. |
| */ |
| static void qd_set_password_from_file(const char *password_file, char **password_field, qd_log_source_t *log_source) |
| { |
| if (password_file) { |
| FILE *file = fopen(password_file, "r"); |
| |
| if (file == NULL) { |
| // |
| // The global variable errno (found in <errno.h>) contains information about what went wrong; you can use perror() to print that information as a readable string |
| // |
| qd_log(log_source, QD_LOG_ERROR, "Unable to open password file %s, error: %s", password_file, strerror(errno)); |
| return; |
| } |
| |
| char buffer[200]; |
| |
| int c; |
| int i=0; |
| |
| while (i < 200 - 1) { |
| c = fgetc(file); |
| if (c == EOF || c == '\n') |
| break; |
| buffer[i++] = c; |
| } |
| |
| if (i != 0) { |
| buffer[i] = '\0'; |
| free(*password_field); |
| *password_field = strdup(buffer); |
| } |
| fclose(file); |
| } |
| } |
| |
| /** |
| * Search the list of config_sasl_plugins for an sasl-profile that matches the passed in name |
| */ |
| static qd_config_sasl_plugin_t *qd_find_sasl_plugin(qd_connection_manager_t *cm, char *name) |
| { |
| qd_config_sasl_plugin_t *sasl_plugin = DEQ_HEAD(cm->config_sasl_plugins); |
| while (sasl_plugin) { |
| if (strcmp(sasl_plugin->name, name) == 0) |
| return sasl_plugin; |
| sasl_plugin = DEQ_NEXT(sasl_plugin); |
| } |
| |
| return 0; |
| } |
| |
| void qd_server_config_free(qd_server_config_t *cf) |
| { |
| if (!cf) return; |
| free(cf->host); |
| free(cf->port); |
| free(cf->host_port); |
| free(cf->role); |
| if (cf->http_root_dir) free(cf->http_root_dir); |
| if (cf->name) free(cf->name); |
| if (cf->protocol_family) free(cf->protocol_family); |
| if (cf->sasl_username) free(cf->sasl_username); |
| if (cf->sasl_password) free(cf->sasl_password); |
| if (cf->sasl_mechanisms) free(cf->sasl_mechanisms); |
| if (cf->ssl_profile) free(cf->ssl_profile); |
| if (cf->failover_list) qd_failover_list_free(cf->failover_list); |
| if (cf->log_message) free(cf->log_message); |
| |
| if (cf->ssl_certificate_file) free(cf->ssl_certificate_file); |
| if (cf->ssl_private_key_file) free(cf->ssl_private_key_file); |
| if (cf->ssl_ciphers) free(cf->ssl_ciphers); |
| if (cf->ssl_protocols) free(cf->ssl_protocols); |
| if (cf->ssl_password) free(cf->ssl_password); |
| if (cf->ssl_trusted_certificate_db) free(cf->ssl_trusted_certificate_db); |
| if (cf->ssl_uid_format) free(cf->ssl_uid_format); |
| if (cf->ssl_uid_name_mapping_file) free(cf->ssl_uid_name_mapping_file); |
| |
| if (cf->sasl_plugin_config.auth_service) free(cf->sasl_plugin_config.auth_service); |
| if (cf->sasl_plugin_config.hostname) free(cf->sasl_plugin_config.hostname); |
| if (cf->sasl_plugin_config.sasl_init_hostname) free(cf->sasl_plugin_config.sasl_init_hostname); |
| if (cf->sasl_plugin_config.ssl_certificate_file) free(cf->sasl_plugin_config.ssl_certificate_file); |
| if (cf->sasl_plugin_config.ssl_private_key_file) free(cf->sasl_plugin_config.ssl_private_key_file); |
| if (cf->sasl_plugin_config.ssl_ciphers) free(cf->sasl_plugin_config.ssl_ciphers); |
| if (cf->sasl_plugin_config.ssl_protocols) free(cf->sasl_plugin_config.ssl_protocols); |
| if (cf->sasl_plugin_config.ssl_password) free(cf->sasl_plugin_config.ssl_password); |
| if (cf->sasl_plugin_config.ssl_trusted_certificate_db) free(cf->sasl_plugin_config.ssl_trusted_certificate_db); |
| if (cf->sasl_plugin_config.ssl_uid_format) free(cf->sasl_plugin_config.ssl_uid_format); |
| if (cf->sasl_plugin_config.ssl_uid_name_mapping_file) free(cf->sasl_plugin_config.ssl_uid_name_mapping_file); |
| |
| memset(cf, 0, sizeof(*cf)); |
| } |
| |
| #define CHECK() if (qd_error_code()) goto error |
| #define SSTRDUP(S) ((S) ? strdup(S) : NULL) |
| |
| /** |
| * Private function to set the values of booleans strip_inbound_annotations and strip_outbound_annotations |
| * based on the corresponding values for the settings in qdrouter.json |
| * strip_inbound_annotations and strip_outbound_annotations are defaulted to true |
| */ |
| static void load_strip_annotations(qd_server_config_t *config, const char* stripAnnotations) |
| { |
| if (stripAnnotations) { |
| if (strcmp(stripAnnotations, "both") == 0) { |
| config->strip_inbound_annotations = true; |
| config->strip_outbound_annotations = true; |
| } |
| else if (strcmp(stripAnnotations, "in") == 0) { |
| config->strip_inbound_annotations = true; |
| config->strip_outbound_annotations = false; |
| } |
| else if (strcmp(stripAnnotations, "out") == 0) { |
| config->strip_inbound_annotations = false; |
| config->strip_outbound_annotations = true; |
| } |
| else if (strcmp(stripAnnotations, "no") == 0) { |
| config->strip_inbound_annotations = false; |
| config->strip_outbound_annotations = false; |
| } |
| } |
| else { |
| assert(stripAnnotations); |
| //This is just for safety. Default to stripInboundAnnotations and stripOutboundAnnotations to true (to "both"). |
| config->strip_inbound_annotations = true; |
| config->strip_outbound_annotations = true; |
| } |
| } |
| |
| /** |
| * Since both the host and the addr have defaults of 127.0.0.1, we will have to use the non-default wherever it is available. |
| */ |
| static void set_config_host(qd_server_config_t *config, qd_entity_t* entity) |
| { |
| config->host = qd_entity_opt_string(entity, "host", 0); |
| |
| assert(config->host); |
| |
| int hplen = strlen(config->host) + strlen(config->port) + 2; |
| config->host_port = malloc(hplen); |
| snprintf(config->host_port, hplen, "%s:%s", config->host, config->port); |
| } |
| |
| static void qd_config_process_password(char **actual_val, char *pw, bool *is_file, bool allow_literal_prefix, qd_log_source_t *log_source) |
| { |
| if (!pw) |
| return; |
| |
| // |
| // If the "password" starts with "env:" then the remaining |
| // text is the environment variable that contains the password |
| // |
| if (strncmp(pw, "env:", 4) == 0) { |
| char *env = pw + 4; |
| // skip the leading whitespace if it is there |
| while (*env == ' ') ++env; |
| |
| const char* passwd = getenv(env); |
| if (passwd) { |
| // |
| // Replace the allocated directive with the looked-up password |
| // |
| *actual_val = strdup(passwd); |
| } else { |
| qd_error(QD_ERROR_NOT_FOUND, "Failed to find a password in the environment variable"); |
| } |
| } |
| |
| // |
| // If the "password" starts with "literal:" or "pass:" then |
| // the remaining text is the password and the heading should be |
| // stripped off |
| // |
| else if ( (strncmp(pw, "literal:", 8) == 0 && allow_literal_prefix) || strncmp(pw, "pass:", 5) == 0) { |
| qd_log(log_source, QD_LOG_WARNING, "It is unsafe to provide plain text passwords in the config file"); |
| |
| if (strncmp(pw, "l", 1) == 0) { |
| // skip the "literal:" header |
| pw += 8; |
| } |
| else { |
| // skip the "pass:" header |
| pw += 5; |
| } |
| |
| // |
| // Replace the password with a copy of the string after "literal: or "pass:" |
| // |
| char *copy = strdup(pw); |
| *actual_val = copy; |
| } |
| // |
| // If the password starts with a file: literal set the is_file to true. |
| // |
| else if (strncmp(pw, "file:", 5) == 0) { |
| pw += 5; |
| |
| // Replace the password with a copy of the string after "file:" |
| char *copy = strdup(pw); |
| *actual_val = copy; |
| *is_file = true; |
| } |
| else { |
| // |
| // THe password field does not have any prefixes. Use it as plain text |
| // |
| qd_log(log_source, QD_LOG_WARNING, "It is unsafe to provide plain text passwords in the config file"); |
| |
| } |
| } |
| |
| static qd_log_bits populate_log_message(const qd_server_config_t *config) |
| { |
| //May have to copy this string since strtok modifies original string. |
| char *log_message = config->log_message; |
| |
| int32_t ret_val = 0; |
| |
| if (!log_message || strcmp(log_message, NONE) == 0) |
| return ret_val; |
| |
| //If log_message is set to 'all', turn on all bits. |
| if (strcmp(log_message, ALL) == 0) |
| return INT32_MAX; |
| |
| char *delim = ","; |
| |
| /* get the first token */ |
| char *token = strtok(log_message, delim); |
| |
| const char *component = 0; |
| |
| /* walk through other tokens */ |
| while( token != NULL ) { |
| for (int i=0;; i++) { |
| component = qd_log_message_components[i]; |
| if (component == 0) |
| break; |
| |
| if (strcmp(component, token) == 0) { |
| ret_val |= 1 << i; |
| } |
| } |
| token = strtok(NULL, delim); |
| } |
| |
| return ret_val; |
| } |
| |
| |
| static qd_error_t load_server_config(qd_dispatch_t *qd, qd_server_config_t *config, qd_entity_t* entity, bool is_listener) |
| { |
| qd_error_clear(); |
| |
| bool authenticatePeer = qd_entity_opt_bool(entity, "authenticatePeer", false); CHECK(); |
| bool verifyHostName = qd_entity_opt_bool(entity, "verifyHostname", true); CHECK(); |
| bool requireEncryption = qd_entity_opt_bool(entity, "requireEncryption", false); CHECK(); |
| bool requireSsl = qd_entity_opt_bool(entity, "requireSsl", false); CHECK(); |
| |
| ZERO(config); |
| config->log_message = qd_entity_opt_string(entity, "messageLoggingComponents", 0); CHECK(); |
| config->log_bits = populate_log_message(config); |
| config->port = qd_entity_get_string(entity, "port"); CHECK(); |
| config->name = qd_entity_opt_string(entity, "name", 0); CHECK(); |
| config->role = qd_entity_get_string(entity, "role"); CHECK(); |
| config->inter_router_cost = qd_entity_opt_long(entity, "cost", 1); CHECK(); |
| config->protocol_family = qd_entity_opt_string(entity, "protocolFamily", 0); CHECK(); |
| config->healthz = qd_entity_opt_bool(entity, "healthz", true); CHECK(); |
| config->metrics = qd_entity_opt_bool(entity, "metrics", true); CHECK(); |
| config->websockets = qd_entity_opt_bool(entity, "websockets", true); CHECK(); |
| config->http = qd_entity_opt_bool(entity, "http", false); CHECK(); |
| config->http_root_dir = qd_entity_opt_string(entity, "httpRootDir", false); CHECK(); |
| config->http = config->http || config->http_root_dir; /* httpRoot implies http */ |
| config->max_frame_size = qd_entity_get_long(entity, "maxFrameSize"); CHECK(); |
| config->max_sessions = qd_entity_get_long(entity, "maxSessions"); CHECK(); |
| uint64_t ssn_frames = qd_entity_opt_long(entity, "maxSessionFrames", 0); CHECK(); |
| config->idle_timeout_seconds = qd_entity_get_long(entity, "idleTimeoutSeconds"); CHECK(); |
| if (is_listener) { |
| config->initial_handshake_timeout_seconds = qd_entity_get_long(entity, "initialHandshakeTimeoutSeconds"); CHECK(); |
| } |
| config->sasl_username = qd_entity_opt_string(entity, "saslUsername", 0); CHECK(); |
| config->sasl_password = qd_entity_opt_string(entity, "saslPassword", 0); CHECK(); |
| config->sasl_mechanisms = qd_entity_opt_string(entity, "saslMechanisms", 0); CHECK(); |
| config->ssl_profile = qd_entity_opt_string(entity, "sslProfile", 0); CHECK(); |
| config->sasl_plugin = qd_entity_opt_string(entity, "saslPlugin", 0); CHECK(); |
| config->link_capacity = qd_entity_opt_long(entity, "linkCapacity", 0); CHECK(); |
| config->multi_tenant = qd_entity_opt_bool(entity, "multiTenant", false); CHECK(); |
| config->policy_vhost = qd_entity_opt_string(entity, "policyVhost", 0); CHECK(); |
| config->conn_props = qd_entity_opt_map(entity, "openProperties"); CHECK(); |
| |
| char *unused = qd_entity_opt_string(entity, "trustedCertsFile", 0); |
| if (unused) { |
| qd_log(qd->connection_manager->log_source, QD_LOG_WARNING, |
| "Configuration listener attribute 'trustedCertsFile' is not used. Specify sslProfile caCertFile instead."); |
| free(unused); |
| } |
| |
| set_config_host(config, entity); |
| |
| if (config->sasl_password) { |
| // |
| //Process the sasl password field and set the right values based on prefixes. |
| // |
| char *actual_pass = 0; |
| bool is_file_path = 0; |
| qd_config_process_password(&actual_pass, config->sasl_password, &is_file_path, false, qd->connection_manager->log_source); |
| if (actual_pass) { |
| if (is_file_path) { |
| qd_set_password_from_file(actual_pass, &config->sasl_password, qd->connection_manager->log_source); |
| free(actual_pass); |
| } |
| else { |
| free(config->sasl_password); |
| config->sasl_password = actual_pass; |
| } |
| } |
| } |
| |
| // |
| // Handle the defaults for various settings |
| // |
| if (config->link_capacity == 0) |
| config->link_capacity = 250; |
| |
| if (config->max_sessions == 0 || config->max_sessions > 32768) |
| // Proton disallows > 32768 |
| config->max_sessions = 32768; |
| |
| if (config->max_frame_size < QD_AMQP_MIN_MAX_FRAME_SIZE) |
| // Silently promote the minimum max-frame-size |
| // Proton will do this but the number is needed for the |
| // incoming capacity calculation. |
| config->max_frame_size = QD_AMQP_MIN_MAX_FRAME_SIZE; |
| |
| // |
| // Given session frame count and max frame size, compute session incoming_capacity |
| // On 64-bit systems the capacity has no practial limit. |
| // On 32-bit systems the largest default capacity is half the process address space. |
| // |
| bool is_64bit = sizeof(size_t) == 8; |
| #define MAX_32BIT_CAPACITY ((size_t)(2147483647)) |
| if (ssn_frames == 0) { |
| config->incoming_capacity = is_64bit ? MAX_32BIT_CAPACITY * (size_t)config->max_frame_size : MAX_32BIT_CAPACITY; |
| } else { |
| // Limited incoming frames. |
| if (is_64bit) { |
| // Specify this to proton by setting capacity to be |
| // the product (max_frame_size * ssn_frames). |
| config->incoming_capacity = (size_t)config->max_frame_size * (size_t)ssn_frames; |
| } else { |
| // 32-bit systems have an upper bound to the capacity |
| uint64_t max_32bit_capacity = (uint64_t)MAX_32BIT_CAPACITY; |
| uint64_t capacity = (uint64_t)config->max_frame_size * (uint64_t)ssn_frames; |
| if (capacity <= max_32bit_capacity) { |
| config->incoming_capacity = (size_t)capacity; |
| } else { |
| config->incoming_capacity = MAX_32BIT_CAPACITY; |
| uint64_t actual_frames = max_32bit_capacity / (uint64_t)config->max_frame_size; |
| |
| qd_log(qd->connection_manager->log_source, QD_LOG_WARNING, |
| "Server configuation for I/O adapter entity name:'%s', host:'%s', port:'%s', " |
| "requested maxSessionFrames truncated from %"PRId64" to %"PRId64, |
| config->name, config->host, config->port, ssn_frames, actual_frames); |
| } |
| } |
| } |
| |
| // |
| // For now we are hardwiring this attribute to true. If there's an outcry from the |
| // user community, we can revisit this later. |
| // |
| config->allowInsecureAuthentication = true; |
| config->verify_host_name = verifyHostName; |
| |
| char *stripAnnotations = qd_entity_opt_string(entity, "stripAnnotations", 0); |
| load_strip_annotations(config, stripAnnotations); |
| free(stripAnnotations); |
| stripAnnotations = 0; |
| CHECK(); |
| |
| config->requireAuthentication = authenticatePeer; |
| config->requireEncryption = requireEncryption || requireSsl; |
| |
| if (config->ssl_profile) { |
| config->ssl_required = requireSsl; |
| config->ssl_require_peer_authentication = config->sasl_mechanisms && |
| strstr(config->sasl_mechanisms, "EXTERNAL") != 0; |
| |
| qd_config_ssl_profile_t *ssl_profile = |
| qd_find_ssl_profile(qd->connection_manager, config->ssl_profile); |
| if (ssl_profile) { |
| config->ssl_certificate_file = SSTRDUP(ssl_profile->ssl_certificate_file); |
| config->ssl_private_key_file = SSTRDUP(ssl_profile->ssl_private_key_file); |
| config->ssl_ciphers = SSTRDUP(ssl_profile->ssl_ciphers); |
| config->ssl_protocols = SSTRDUP(ssl_profile->ssl_protocols); |
| config->ssl_password = SSTRDUP(ssl_profile->ssl_password); |
| config->ssl_trusted_certificate_db = SSTRDUP(ssl_profile->ssl_trusted_certificate_db); |
| config->ssl_uid_format = SSTRDUP(ssl_profile->ssl_uid_format); |
| config->ssl_uid_name_mapping_file = SSTRDUP(ssl_profile->uid_name_mapping_file); |
| } |
| } |
| |
| if (config->sasl_plugin) { |
| qd_config_sasl_plugin_t *sasl_plugin = |
| qd_find_sasl_plugin(qd->connection_manager, config->sasl_plugin); |
| if (sasl_plugin) { |
| config->sasl_plugin_config.auth_service = SSTRDUP(sasl_plugin->auth_service); |
| config->sasl_plugin_config.hostname = SSTRDUP(sasl_plugin->hostname); |
| config->sasl_plugin_config.sasl_init_hostname = SSTRDUP(sasl_plugin->sasl_init_hostname); |
| qd_log(qd->connection_manager->log_source, QD_LOG_INFO, "Using auth service %s from SASL Plugin %s", config->sasl_plugin_config.auth_service, config->sasl_plugin); |
| |
| if (sasl_plugin->auth_ssl_profile) { |
| config->sasl_plugin_config.use_ssl = true; |
| qd_config_ssl_profile_t *auth_ssl_profile = |
| qd_find_ssl_profile(qd->connection_manager, sasl_plugin->auth_ssl_profile); |
| |
| config->sasl_plugin_config.ssl_certificate_file = SSTRDUP(auth_ssl_profile->ssl_certificate_file); |
| config->sasl_plugin_config.ssl_private_key_file = SSTRDUP(auth_ssl_profile->ssl_private_key_file); |
| config->sasl_plugin_config.ssl_ciphers = SSTRDUP(auth_ssl_profile->ssl_ciphers); |
| config->sasl_plugin_config.ssl_protocols = SSTRDUP(auth_ssl_profile->ssl_protocols); |
| config->sasl_plugin_config.ssl_password = SSTRDUP(auth_ssl_profile->ssl_password); |
| config->sasl_plugin_config.ssl_trusted_certificate_db = SSTRDUP(auth_ssl_profile->ssl_trusted_certificate_db); |
| config->sasl_plugin_config.ssl_uid_format = SSTRDUP(auth_ssl_profile->ssl_uid_format); |
| config->sasl_plugin_config.ssl_uid_name_mapping_file = SSTRDUP(auth_ssl_profile->uid_name_mapping_file); |
| } else { |
| config->sasl_plugin_config.use_ssl = false; |
| } |
| } else { |
| qd_error(QD_ERROR_RUNTIME, "Cannot find sasl plugin %s", config->sasl_plugin); CHECK(); |
| } |
| } |
| |
| return QD_ERROR_NONE; |
| |
| error: |
| qd_server_config_free(config); |
| return qd_error_code(); |
| } |
| |
| |
| bool is_log_component_enabled(qd_log_bits log_message, const char *component_name) { |
| |
| for(int i=0;;i++) { |
| const char *component = qd_log_message_components[i]; |
| if (component == 0) |
| break; |
| if (strcmp(component_name, component) == 0) |
| return (log_message >> i) & 1; |
| } |
| |
| return 0; |
| } |
| |
| |
| static bool config_ssl_profile_free(qd_connection_manager_t *cm, qd_config_ssl_profile_t *ssl_profile) |
| { |
| DEQ_REMOVE(cm->config_ssl_profiles, ssl_profile); |
| |
| free(ssl_profile->name); |
| free(ssl_profile->ssl_password); |
| free(ssl_profile->ssl_trusted_certificate_db); |
| free(ssl_profile->ssl_uid_format); |
| free(ssl_profile->uid_name_mapping_file); |
| free(ssl_profile->ssl_certificate_file); |
| free(ssl_profile->ssl_private_key_file); |
| free(ssl_profile->ssl_ciphers); |
| free(ssl_profile->ssl_protocols); |
| free(ssl_profile); |
| return true; |
| |
| } |
| |
| static bool config_sasl_plugin_free(qd_connection_manager_t *cm, qd_config_sasl_plugin_t *sasl_plugin) |
| { |
| DEQ_REMOVE(cm->config_sasl_plugins, sasl_plugin); |
| |
| free(sasl_plugin->name); |
| free(sasl_plugin->auth_service); |
| free(sasl_plugin->hostname); |
| free(sasl_plugin->sasl_init_hostname); |
| free(sasl_plugin->auth_ssl_profile); |
| free(sasl_plugin); |
| return true; |
| |
| } |
| |
| |
| QD_EXPORT qd_config_ssl_profile_t *qd_dispatch_configure_ssl_profile(qd_dispatch_t *qd, qd_entity_t *entity) |
| { |
| qd_error_clear(); |
| qd_connection_manager_t *cm = qd->connection_manager; |
| |
| qd_config_ssl_profile_t *ssl_profile = NEW(qd_config_ssl_profile_t); |
| ZERO(ssl_profile); |
| DEQ_ITEM_INIT(ssl_profile); |
| DEQ_INSERT_TAIL(cm->config_ssl_profiles, ssl_profile); |
| ssl_profile->name = qd_entity_opt_string(entity, "name", 0); CHECK(); |
| ssl_profile->ssl_certificate_file = qd_entity_opt_string(entity, "certFile", 0); CHECK(); |
| ssl_profile->ssl_private_key_file = qd_entity_opt_string(entity, "privateKeyFile", 0); CHECK(); |
| ssl_profile->ssl_password = qd_entity_opt_string(entity, "password", 0); CHECK(); |
| char *password_file = qd_entity_opt_string(entity, "passwordFile", 0); CHECK(); |
| |
| if (ssl_profile->ssl_password) { |
| // |
| // Process the password to handle any modifications or lookups needed |
| // |
| char *actual_pass = 0; |
| bool is_file_path = 0; |
| qd_config_process_password(&actual_pass, ssl_profile->ssl_password, &is_file_path, true, cm->log_source); CHECK(); |
| if (actual_pass) { |
| if (is_file_path) { |
| qd_set_password_from_file(actual_pass, &ssl_profile->ssl_password, cm->log_source); |
| free(actual_pass); |
| } |
| else { |
| free(ssl_profile->ssl_password); |
| ssl_profile->ssl_password = actual_pass; |
| } |
| } |
| } |
| else if (password_file) { |
| // |
| // Warn the user that the passwordFile attribute has been deprecated. |
| // |
| qd_log(cm->log_source, QD_LOG_WARNING, "Attribute passwordFile of entity sslProfile has been deprecated. Use password field with the file: prefix instead."); |
| qd_set_password_from_file(password_file, &ssl_profile->ssl_password, cm->log_source); |
| } |
| |
| free(password_file); |
| |
| ssl_profile->ssl_ciphers = qd_entity_opt_string(entity, "ciphers", 0); CHECK(); |
| ssl_profile->ssl_protocols = qd_entity_opt_string(entity, "protocols", 0); CHECK(); |
| ssl_profile->ssl_trusted_certificate_db = qd_entity_opt_string(entity, "caCertFile", 0); CHECK(); |
| ssl_profile->ssl_uid_format = qd_entity_opt_string(entity, "uidFormat", 0); CHECK(); |
| ssl_profile->uid_name_mapping_file = qd_entity_opt_string(entity, "uidNameMappingFile", 0); CHECK(); |
| |
| qd_log(cm->log_source, QD_LOG_INFO, "Created SSL Profile with name %s ", ssl_profile->name); |
| return ssl_profile; |
| |
| error: |
| qd_log(cm->log_source, QD_LOG_ERROR, "Unable to create ssl profile: %s", qd_error_message()); |
| config_ssl_profile_free(cm, ssl_profile); |
| return 0; |
| } |
| |
| QD_EXPORT qd_config_sasl_plugin_t *qd_dispatch_configure_sasl_plugin(qd_dispatch_t *qd, qd_entity_t *entity) |
| { |
| qd_error_clear(); |
| qd_connection_manager_t *cm = qd->connection_manager; |
| |
| qd_config_sasl_plugin_t *sasl_plugin = NEW(qd_config_sasl_plugin_t); |
| ZERO(sasl_plugin); |
| DEQ_ITEM_INIT(sasl_plugin); |
| DEQ_INSERT_TAIL(cm->config_sasl_plugins, sasl_plugin); |
| sasl_plugin->name = qd_entity_opt_string(entity, "name", 0); CHECK(); |
| |
| sasl_plugin->hostname = qd_entity_opt_string(entity, "host", 0); |
| char *auth_port = qd_entity_opt_string(entity, "port", 0); |
| |
| if (sasl_plugin->hostname && auth_port) { |
| int strlen_auth_host = strlen(sasl_plugin->hostname); |
| int strlen_auth_port = strlen(auth_port); |
| |
| if (strlen_auth_host > 0 && strlen_auth_port > 0) { |
| |
| int hplen = strlen_auth_host + strlen_auth_port + 2; |
| if (hplen > 2) { |
| sasl_plugin->auth_service = malloc(hplen); |
| snprintf(sasl_plugin->auth_service, hplen, "%s:%s", sasl_plugin->hostname, auth_port); |
| } |
| } |
| } |
| |
| free(auth_port); |
| |
| if (!sasl_plugin->auth_service) { |
| sasl_plugin->auth_service = qd_entity_opt_string(entity, "authService", 0); CHECK(); |
| qd_log(cm->log_source, QD_LOG_WARNING, "Attribute authService of entity authServicePlugin has been deprecated. Use host and port instead."); |
| } |
| |
| sasl_plugin->sasl_init_hostname = qd_entity_opt_string(entity, "realm", 0); CHECK(); |
| sasl_plugin->auth_ssl_profile = qd_entity_opt_string(entity, "sslProfile", 0); CHECK(); |
| |
| qd_log(cm->log_source, QD_LOG_INFO, "Created SASL plugin config with name %s", sasl_plugin->name); |
| return sasl_plugin; |
| |
| error: |
| qd_log(cm->log_source, QD_LOG_ERROR, "Unable to create SASL plugin config: %s", qd_error_message()); |
| config_sasl_plugin_free(cm, sasl_plugin); |
| return 0; |
| } |
| |
| static void log_config(qd_log_source_t *log, qd_server_config_t *c, const char *what) { |
| qd_log(log, QD_LOG_INFO, "Configured %s: %s proto=%s, role=%s%s%s%s", |
| what, c->host_port, c->protocol_family ? c->protocol_family : "any", |
| c->role, |
| c->http ? ", http" : "", |
| c->ssl_profile ? ", sslProfile=":"", |
| c->ssl_profile ? c->ssl_profile:""); |
| } |
| |
| |
| QD_EXPORT qd_listener_t *qd_dispatch_configure_listener(qd_dispatch_t *qd, qd_entity_t *entity) |
| { |
| qd_connection_manager_t *cm = qd->connection_manager; |
| qd_listener_t *li = qd_server_listener(qd->server); |
| if (!li || load_server_config(qd, &li->config, entity, true) != QD_ERROR_NONE) { |
| qd_log(cm->log_source, QD_LOG_ERROR, "Unable to create listener: %s", qd_error_message()); |
| qd_listener_decref(li); |
| return 0; |
| } |
| char *fol = qd_entity_opt_string(entity, "failoverUrls", 0); |
| if (fol) { |
| li->config.failover_list = qd_failover_list(fol); |
| free(fol); |
| if (li->config.failover_list == 0) { |
| qd_log(cm->log_source, QD_LOG_ERROR, "Unable to create listener, bad failover list: %s", |
| qd_error_message()); |
| qd_listener_decref(li); |
| return 0; |
| } |
| } else { |
| li->config.failover_list = 0; |
| } |
| DEQ_ITEM_INIT(li); |
| DEQ_INSERT_TAIL(cm->listeners, li); |
| log_config(cm->log_source, &li->config, "Listener"); |
| return li; |
| } |
| |
| |
| QD_EXPORT qd_error_t qd_entity_refresh_listener(qd_entity_t* entity, void *impl) |
| { |
| return QD_ERROR_NONE; |
| } |
| |
| |
| /** |
| * Calculates the total length of the failover list string. |
| * For example, the failover list string can look like this - "amqp://0.0.0.0:62616, amqp://0.0.0.0:61616" |
| * This function calculates the length of the above string by adding up the scheme (amqp or amqps) and host_port for each failover item. |
| * It also assumes that there will be a comma and a space between each failover item. |
| * |
| */ |
| static int get_failover_info_length(qd_failover_item_list_t conn_info_list) |
| { |
| int arr_length = 0; |
| qd_failover_item_t *item = DEQ_HEAD(conn_info_list); |
| |
| while(item) { |
| if (item->scheme) { |
| // The +3 is for the '://' |
| arr_length += strlen(item->scheme) + 3; |
| } |
| if (item->host_port) { |
| arr_length += strlen(item->host_port); |
| } |
| item = DEQ_NEXT(item); |
| if (item) { |
| // This is for the comma and space between the items |
| arr_length += 2; |
| } |
| } |
| |
| if (arr_length > 0) |
| // This is for the final '\0' |
| arr_length += 1; |
| |
| return arr_length; |
| } |
| |
| /** |
| * |
| * Creates a failover url list. This comma separated failover list shows a list of urls that the router will attempt |
| * to connect to in case the primary connection fails. The router will attempt these failover connections to urls in |
| * the order that they appear in the list. |
| * |
| */ |
| qd_error_t qd_entity_refresh_connector(qd_entity_t* entity, void *impl) |
| { |
| qd_connector_t *connector = (qd_connector_t*) impl; |
| |
| int conn_index = connector->conn_index; |
| |
| int i = 1; |
| int num_items = 0; |
| |
| sys_mutex_lock(connector->lock); |
| |
| qd_failover_item_list_t conn_info_list = connector->conn_info_list; |
| |
| int conn_info_len = DEQ_SIZE(conn_info_list); |
| |
| qd_failover_item_t *item = DEQ_HEAD(conn_info_list); |
| |
| int arr_length = get_failover_info_length(conn_info_list); |
| |
| // This is the string that will contain the comma separated failover list |
| char *failover_info = qd_calloc(arr_length + 1, sizeof(char)); |
| while(item) { |
| |
| // Break out of the loop when we have hit all items in the list. |
| if (num_items >= conn_info_len) |
| break; |
| |
| if (num_items >= 1) { |
| strcat(failover_info, ", "); |
| } |
| |
| // We need to go to the elements in the list to get to the |
| // element that matches the connection index. This is the first |
| // url that the router will try to connect on failover. |
| if (conn_index == i) { |
| num_items += 1; |
| if (item->scheme) { |
| strcat(failover_info, item->scheme); |
| strcat(failover_info, "://"); |
| } |
| if (item->host_port) { |
| strcat(failover_info, item->host_port); |
| } |
| } |
| else { |
| if (num_items > 0) { |
| num_items += 1; |
| if (item->scheme) { |
| strcat(failover_info, item->scheme); |
| strcat(failover_info, "://"); |
| } |
| if (item->host_port) { |
| strcat(failover_info, item->host_port); |
| } |
| } |
| } |
| |
| i += 1; |
| |
| item = DEQ_NEXT(item); |
| if (item == 0) |
| item = DEQ_HEAD(conn_info_list); |
| } |
| |
| const char *state_info = 0; |
| switch (connector->state) { |
| case CXTR_STATE_CONNECTING: |
| state_info = "CONNECTING"; |
| break; |
| case CXTR_STATE_OPEN: |
| state_info = "SUCCESS"; |
| break; |
| case CXTR_STATE_FAILED: |
| state_info = "FAILED"; |
| break; |
| case CXTR_STATE_INIT: |
| state_info = "INITIALIZING"; |
| break; |
| case CXTR_STATE_DELETED: |
| // deleted by management, waiting for connection to close |
| state_info = "CLOSING"; |
| break; |
| default: |
| state_info = "UNKNOWN"; |
| break; |
| } |
| |
| if (qd_entity_set_string(entity, "failoverUrls", failover_info) == 0 |
| && qd_entity_set_string(entity, "connectionStatus", state_info) == 0 |
| && qd_entity_set_string(entity, "connectionMsg", connector->conn_msg) == 0) { |
| |
| sys_mutex_unlock(connector->lock); |
| free(failover_info); |
| return QD_ERROR_NONE; |
| } |
| |
| sys_mutex_unlock(connector->lock); |
| free(failover_info); |
| return qd_error_code(); |
| } |
| |
| |
| QD_EXPORT qd_connector_t *qd_dispatch_configure_connector(qd_dispatch_t *qd, qd_entity_t *entity) |
| { |
| qd_connection_manager_t *cm = qd->connection_manager; |
| qd_connector_t *ct = qd_server_connector(qd->server); |
| |
| qd_error_clear(); |
| |
| if (ct && load_server_config(qd, &ct->config, entity, false) == QD_ERROR_NONE) { |
| ct->policy_vhost = qd_entity_opt_string(entity, "policyVhost", 0); CHECK(); |
| DEQ_ITEM_INIT(ct); |
| DEQ_INSERT_TAIL(cm->connectors, ct); |
| log_config(cm->log_source, &ct->config, "Connector"); |
| |
| // |
| // Add the first item to the ct->conn_info_list |
| // The initial connection information and any backup connection information is stored in the conn_info_list |
| // |
| qd_failover_item_t *item = NEW(qd_failover_item_t); |
| ZERO(item); |
| if (ct->config.ssl_required) |
| item->scheme = strdup("amqps"); |
| else |
| item->scheme = strdup("amqp"); |
| |
| item->host = strdup(ct->config.host); |
| item->port = strdup(ct->config.port); |
| |
| int hplen = strlen(item->host) + strlen(item->port) + 2; |
| item->host_port = malloc(hplen); |
| snprintf(item->host_port, hplen, "%s:%s", item->host , item->port); |
| |
| DEQ_INSERT_TAIL(ct->conn_info_list, item); |
| |
| return ct; |
| } |
| |
| error: |
| qd_log(cm->log_source, QD_LOG_ERROR, "Unable to create connector: %s", qd_error_message()); |
| qd_connector_decref(ct); |
| return 0; |
| } |
| |
| |
| qd_connection_manager_t *qd_connection_manager(qd_dispatch_t *qd) |
| { |
| qd_connection_manager_t *cm = NEW(qd_connection_manager_t); |
| if (!cm) |
| return 0; |
| |
| cm->log_source = qd_log_source("CONN_MGR"); |
| cm->server = qd->server; |
| DEQ_INIT(cm->listeners); |
| DEQ_INIT(cm->connectors); |
| DEQ_INIT(cm->config_ssl_profiles); |
| DEQ_INIT(cm->config_sasl_plugins); |
| |
| return cm; |
| } |
| |
| |
| // Called on router shutdown |
| // |
| void qd_connection_manager_free(qd_connection_manager_t *cm) |
| { |
| if (!cm) return; |
| qd_listener_t *li = DEQ_HEAD(cm->listeners); |
| while (li) { |
| DEQ_REMOVE_HEAD(cm->listeners); |
| if (li->pn_listener) { |
| // DISPATCH-1508: force cleanup of pn_listener context. This is |
| // usually done in the PN_LISTENER_CLOSE event handler in server.c, |
| // but since the router is going down those events will no longer |
| // be generated. |
| pn_listener_set_context(li->pn_listener, 0); |
| pn_listener_close(li->pn_listener); |
| li->pn_listener = 0; |
| qd_listener_decref(li); // for the pn_listener's context |
| } |
| qd_listener_decref(li); |
| li = DEQ_HEAD(cm->listeners); |
| } |
| |
| qd_connector_t *connector = DEQ_HEAD(cm->connectors); |
| while (connector) { |
| DEQ_REMOVE_HEAD(cm->connectors); |
| sys_mutex_lock(connector->lock); |
| // setting DELETED below ensures the timer callback |
| // will not initiate a re-connect once we drop |
| // the lock |
| connector->state = CXTR_STATE_DELETED; |
| sys_mutex_unlock(connector->lock); |
| // cannot cancel timer while holding lock since the |
| // callback takes the lock |
| qd_timer_cancel(connector->timer); |
| qd_connector_decref(connector); |
| |
| connector = DEQ_HEAD(cm->connectors); |
| } |
| |
| qd_config_ssl_profile_t *sslp = DEQ_HEAD(cm->config_ssl_profiles); |
| while (sslp) { |
| config_ssl_profile_free(cm, sslp); |
| sslp = DEQ_HEAD(cm->config_ssl_profiles); |
| } |
| |
| qd_config_sasl_plugin_t *saslp = DEQ_HEAD(cm->config_sasl_plugins); |
| while (saslp) { |
| config_sasl_plugin_free(cm, saslp); |
| saslp = DEQ_HEAD(cm->config_sasl_plugins); |
| } |
| |
| free(cm); |
| } |
| |
| |
| /** NOTE: non-static qd_connection_manager_* functions are called from the python agent */ |
| |
| |
| void qd_connection_manager_start(qd_dispatch_t *qd) |
| { |
| static bool first_start = true; |
| qd_listener_t *li = DEQ_HEAD(qd->connection_manager->listeners); |
| qd_connector_t *ct = DEQ_HEAD(qd->connection_manager->connectors); |
| |
| while (li) { |
| if (!li->pn_listener) { |
| if (!qd_listener_listen(li) && first_start) { |
| qd_log(qd->connection_manager->log_source, QD_LOG_CRITICAL, |
| "Listen on %s failed during initial config", li->config.host_port); |
| exit(1); |
| } else { |
| li->exit_on_error = first_start; |
| } |
| } |
| li = DEQ_NEXT(li); |
| } |
| |
| while (ct) { |
| |
| if (ct->state == CXTR_STATE_OPEN || ct->state == CXTR_STATE_CONNECTING) { |
| ct = DEQ_NEXT(ct); |
| continue; |
| } |
| |
| qd_connector_connect(ct); |
| ct = DEQ_NEXT(ct); |
| } |
| |
| first_start = false; |
| } |
| |
| |
| QD_EXPORT void qd_connection_manager_delete_listener(qd_dispatch_t *qd, void *impl) |
| { |
| qd_listener_t *li = (qd_listener_t*) impl; |
| if (li) { |
| if (li->pn_listener) { |
| pn_listener_close(li->pn_listener); |
| } |
| else if (li->http) { |
| qd_lws_listener_close(li->http); |
| } |
| DEQ_REMOVE(qd->connection_manager->listeners, li); |
| qd_listener_decref(li); |
| } |
| } |
| |
| |
| QD_EXPORT void qd_connection_manager_delete_ssl_profile(qd_dispatch_t *qd, void *impl) |
| { |
| qd_config_ssl_profile_t *ssl_profile = (qd_config_ssl_profile_t*) impl; |
| config_ssl_profile_free(qd->connection_manager, ssl_profile); |
| } |
| |
| QD_EXPORT void qd_connection_manager_delete_sasl_plugin(qd_dispatch_t *qd, void *impl) |
| { |
| qd_config_sasl_plugin_t *sasl_plugin = (qd_config_sasl_plugin_t*) impl; |
| config_sasl_plugin_free(qd->connection_manager, sasl_plugin); |
| } |
| |
| |
| static void deferred_close(void *context, bool discard) { |
| if (!discard) { |
| pn_connection_close((pn_connection_t*)context); |
| } |
| } |
| |
| |
| // threading: called by management thread while I/O thread may be |
| // referencing the qd_connector_t via the qd_connection_t |
| // |
| QD_EXPORT void qd_connection_manager_delete_connector(qd_dispatch_t *qd, void *impl) |
| { |
| qd_connector_t *ct = (qd_connector_t*) impl; |
| if (ct) { |
| // cannot free the timer while holding ct->lock since the |
| // timer callback may be running during the call to qd_timer_free |
| qd_timer_t *timer = 0; |
| void *dct = qd_connection_new_qd_deferred_call_t(); |
| sys_mutex_lock(ct->lock); |
| timer = ct->timer; |
| ct->timer = 0; |
| ct->state = CXTR_STATE_DELETED; |
| qd_connection_t *conn = ct->qd_conn; |
| if (conn && conn->pn_conn) { |
| qd_connection_invoke_deferred_impl(conn, deferred_close, conn->pn_conn, dct); |
| sys_mutex_unlock(ct->lock); |
| } else { |
| sys_mutex_unlock(ct->lock); |
| qd_connection_free_qd_deferred_call_t(dct); |
| } |
| qd_timer_free(timer); |
| DEQ_REMOVE(qd->connection_manager->connectors, ct); |
| qd_connector_decref(ct); |
| } |
| } |
| |
| |
| const char *qd_connector_name(qd_connector_t *ct) |
| { |
| return ct ? ct->config.name : 0; |
| } |
| |
