22 #include <sys/param.h> 24 #include <sys/types.h> 27 #include <sys/socket.h> 28 #include <arpa/inet.h> 29 #include <netinet/in.h> 30 #include <netinet/ip.h> 31 #include <netinet/tcp.h> 45 #ifdef HAVE_GNUTLS_GNUTLS_H 47 # include <gnutls/gnutls.h> 49 const int psk_tls_kx_order[] = {
54 const int anon_tls_kx_order[] = {
64 #ifdef HAVE_LINUX_SWAB_H 65 # include <linux/swab.h> 71 #define __swab16(x) ((uint16_t)( \ 72 (((uint16_t)(x) & (uint16_t)0x00ffU) << 8) | \ 73 (((uint16_t)(x) & (uint16_t)0xff00U) >> 8))) 75 #define __swab32(x) ((uint32_t)( \ 76 (((uint32_t)(x) & (uint32_t)0x000000ffUL) << 24) | \ 77 (((uint32_t)(x) & (uint32_t)0x0000ff00UL) << 8) | \ 78 (((uint32_t)(x) & (uint32_t)0x00ff0000UL) >> 8) | \ 79 (((uint32_t)(x) & (uint32_t)0xff000000UL) >> 24))) 81 #define __swab64(x) ((uint64_t)( \ 82 (((uint64_t)(x) & (uint64_t)0x00000000000000ffULL) << 56) | \ 83 (((uint64_t)(x) & (uint64_t)0x000000000000ff00ULL) << 40) | \ 84 (((uint64_t)(x) & (uint64_t)0x0000000000ff0000ULL) << 24) | \ 85 (((uint64_t)(x) & (uint64_t)0x00000000ff000000ULL) << 8) | \ 86 (((uint64_t)(x) & (uint64_t)0x000000ff00000000ULL) >> 8) | \ 87 (((uint64_t)(x) & (uint64_t)0x0000ff0000000000ULL) >> 24) | \ 88 (((uint64_t)(x) & (uint64_t)0x00ff000000000000ULL) >> 40) | \ 89 (((uint64_t)(x) & (uint64_t)0xff00000000000000ULL) >> 56))) 92 #define REMOTE_MSG_VERSION 1 93 #define ENDIAN_LOCAL 0xBADADBBD 95 struct crm_remote_header_v0
110 static struct crm_remote_header_v0 *
113 struct crm_remote_header_v0 *header = (
struct crm_remote_header_v0 *)remote->
buffer;
114 if(remote->
buffer_offset <
sizeof(
struct crm_remote_header_v0)) {
122 crm_err(
"Invalid message detected, endian mismatch: %lx is neither %lx nor the swab'd %lx",
128 header->flags =
__swab64(header->flags);
129 header->endian =
__swab32(header->endian);
131 header->version =
__swab32(header->version);
132 header->size_total =
__swab32(header->size_total);
133 header->payload_offset =
__swab32(header->payload_offset);
134 header->payload_compressed =
__swab32(header->payload_compressed);
135 header->payload_uncompressed =
__swab32(header->payload_uncompressed);
141 #ifdef HAVE_GNUTLS_GNUTLS_H 144 crm_initiate_client_tls_handshake(
crm_remote_t * remote,
int timeout_ms)
148 time_t start = time(NULL);
151 rc = gnutls_handshake(*remote->tls_session);
152 if (rc == GNUTLS_E_INTERRUPTED || rc == GNUTLS_E_AGAIN) {
160 }
while (((time(NULL) - start) < (timeout_ms / 1000)) &&
161 (rc == GNUTLS_E_INTERRUPTED || rc == GNUTLS_E_AGAIN));
164 crm_trace(
"gnutls_handshake() failed with %d", rc);
178 const char *dh_min_bits_s = getenv(
"PCMK_dh_min_bits");
187 if (dh_min_bits > 0) {
188 crm_info(
"Requiring server use a Diffie-Hellman prime of at least %d bits",
190 gnutls_dh_set_prime_bits(*session, dh_min_bits);
196 pcmk__bound_dh_bits(
unsigned int dh_bits)
198 const char *dh_min_bits_s = getenv(
"PCMK_dh_min_bits");
199 const char *dh_max_bits_s = getenv(
"PCMK_dh_max_bits");
208 if ((dh_min_bits > 0) && (dh_max_bits > 0)
209 && (dh_max_bits < dh_min_bits)) {
210 crm_warn(
"Ignoring PCMK_dh_max_bits because it is less than PCMK_dh_min_bits");
214 if ((dh_min_bits > 0) && (dh_bits < dh_min_bits)) {
217 if ((dh_max_bits > 0) && (dh_bits > dh_max_bits)) {
235 pcmk__new_tls_session(
int csock,
unsigned int conn_type,
236 gnutls_credentials_type_t cred_type,
void *credentials)
238 int rc = GNUTLS_E_SUCCESS;
239 # ifdef HAVE_GNUTLS_PRIORITY_SET_DIRECT 240 const char *prio_base = NULL;
245 # ifdef HAVE_GNUTLS_PRIORITY_SET_DIRECT 253 prio_base = getenv(
"PCMK_tls_priorities");
254 if (prio_base == NULL) {
258 (cred_type == GNUTLS_CRD_ANON)?
"+ANON-DH" :
"+DHE-PSK:+PSK");
262 if (session == NULL) {
263 rc = GNUTLS_E_MEMORY_ERROR;
267 rc = gnutls_init(session, conn_type);
268 if (rc != GNUTLS_E_SUCCESS) {
272 # ifdef HAVE_GNUTLS_PRIORITY_SET_DIRECT 277 rc = gnutls_priority_set_direct(*session, prio, NULL);
278 if (rc != GNUTLS_E_SUCCESS) {
281 if (conn_type == GNUTLS_CLIENT) {
282 pcmk__set_minimum_dh_bits(session);
285 gnutls_set_default_priority(*session);
286 gnutls_kx_set_priority(*session, (cred_type == GNUTLS_CRD_ANON)? anon_tls_kx_order : psk_tls_kx_order);
289 gnutls_transport_set_ptr(*session,
290 (gnutls_transport_ptr_t) GINT_TO_POINTER(csock));
292 rc = gnutls_credentials_set(*session, cred_type, credentials);
293 if (rc != GNUTLS_E_SUCCESS) {
296 # ifdef HAVE_GNUTLS_PRIORITY_SET_DIRECT 303 # ifdef HAVE_GNUTLS_PRIORITY_SET_DIRECT 304 const char *prio_s = prio;
306 const char *prio_s =
"default";
309 crm_err(
"Could not initialize %s TLS %s session: %s " 310 CRM_XS " rc=%d priority='%s'",
311 (cred_type == GNUTLS_CRD_ANON)?
"anonymous" :
"PSK",
312 (conn_type == GNUTLS_SERVER)?
"server" :
"client",
313 gnutls_strerror(rc), rc, prio_s);
315 # ifdef HAVE_GNUTLS_PRIORITY_SET_DIRECT 318 if (session != NULL) {
319 gnutls_free(session);
340 pcmk__init_tls_dh(gnutls_dh_params_t *dh_params)
342 int rc = GNUTLS_E_SUCCESS;
343 unsigned int dh_bits = 0;
345 rc = gnutls_dh_params_init(dh_params);
346 if (rc != GNUTLS_E_SUCCESS) {
350 #ifdef HAVE_GNUTLS_SEC_PARAM_TO_PK_BITS 351 dh_bits = gnutls_sec_param_to_pk_bits(GNUTLS_PK_DH,
352 GNUTLS_SEC_PARAM_NORMAL);
354 rc = GNUTLS_E_DH_PRIME_UNACCEPTABLE;
360 dh_bits = pcmk__bound_dh_bits(dh_bits);
362 crm_info(
"Generating Diffie-Hellman parameters with %u-bit prime for TLS",
364 rc = gnutls_dh_params_generate2(*dh_params, dh_bits);
365 if (rc != GNUTLS_E_SUCCESS) {
372 crm_err(
"Could not initialize Diffie-Hellman parameters for TLS: %s " 373 CRM_XS " rc=%d", gnutls_strerror(rc), rc);
398 rc = gnutls_handshake(*client->
remote->tls_session);
399 }
while (rc == GNUTLS_E_INTERRUPTED);
401 if (rc == GNUTLS_E_AGAIN) {
406 }
else if (rc != GNUTLS_E_SUCCESS) {
415 const char *unsent = buf;
424 crm_trace(
"Message size: %llu", (
unsigned long long) len);
427 rc = gnutls_record_send(*session, unsent, len);
429 if (rc == GNUTLS_E_INTERRUPTED || rc == GNUTLS_E_AGAIN) {
431 (
unsigned long long) len);
435 gnutls_strerror(rc), rc);
439 }
else if (rc < len) {
440 crm_debug(
"Sent %d of %llu bytes", rc, (
unsigned long long) len);
449 return rc < 0 ? rc : total_send;
454 crm_send_plaintext(
int sock,
const char *buf,
size_t len)
458 const char *unsent = buf;
466 crm_trace(
"Message on socket %d: size=%llu",
467 sock, (
unsigned long long) len);
469 rc = write(sock, unsent, len);
478 crm_perror(LOG_ERR,
"Could only write %d of the remaining %d bytes", rc, (
int)len);
482 }
else if (rc < len) {
483 crm_trace(
"Only sent %d of %llu remaining bytes",
484 rc, (
unsigned long long) len);
490 crm_trace(
"Sent %d bytes: %.100s", rc, buf);
493 return rc < 0 ? rc : total_send;
498 crm_remote_sendv(
crm_remote_t * remote,
struct iovec * iov,
int iovs)
501 int rc = -ESOCKTNOSUPPORT;
503 for(; lpc < iovs; lpc++) {
505 #ifdef HAVE_GNUTLS_GNUTLS_H 506 if (remote->tls_session) {
507 rc = crm_send_tls(remote->tls_session, iov[lpc].iov_base, iov[lpc].iov_len);
512 rc = crm_send_plaintext(remote->
tcp_socket, iov[lpc].iov_base, iov[lpc].iov_len);
515 crm_err(
"Unsupported connection type");
525 static uint64_t
id = 0;
529 struct crm_remote_header_v0 *header;
531 if (xml_text == NULL) {
532 crm_err(
"Could not send remote message: no message provided");
536 header = calloc(1,
sizeof(
struct crm_remote_header_v0));
537 iov[0].iov_base = header;
538 iov[0].iov_len =
sizeof(
struct crm_remote_header_v0);
540 iov[1].iov_base = xml_text;
541 iov[1].iov_len = 1 + strlen(xml_text);
547 header->payload_offset = iov[0].iov_len;
548 header->payload_uncompressed = iov[1].iov_len;
549 header->size_total = iov[0].iov_len + iov[1].iov_len;
552 (
int)iov[0].iov_len, *(
int*)(
void*)xml_text);
553 rc = crm_remote_sendv(remote, iov, 2);
555 crm_err(
"Could not send remote message: %s " CRM_XS " rc=%d",
559 free(iov[0].iov_base);
560 free(iov[1].iov_base);
574 struct crm_remote_header_v0 *header = crm_remote_header(remote);
576 if (remote->
buffer == NULL || header == NULL) {
581 if (header->payload_compressed) {
583 unsigned int size_u = 1 + header->payload_uncompressed;
584 char *uncompressed = calloc(1, header->payload_offset + size_u);
586 crm_trace(
"Decompressing message data %d bytes into %d bytes",
587 header->payload_compressed, size_u);
589 rc = BZ2_bzBuffToBuffDecompress(uncompressed + header->payload_offset, &size_u,
590 remote->
buffer + header->payload_offset,
591 header->payload_compressed, 1, 0);
594 crm_warn(
"Couldn't decompress v%d message, we only understand v%d",
599 }
else if (rc != BZ_OK) {
605 CRM_ASSERT(size_u == header->payload_uncompressed);
607 memcpy(uncompressed, remote->
buffer, header->payload_offset);
608 remote->
buffer_size = header->payload_offset + size_u;
611 remote->
buffer = uncompressed;
612 header = crm_remote_header(remote);
618 CRM_LOG_ASSERT(remote->
buffer[
sizeof(
struct crm_remote_header_v0) + header->payload_uncompressed - 1] == 0);
622 crm_warn(
"Couldn't parse v%d message, we only understand v%d",
625 }
else if (xml == NULL) {
626 crm_err(
"Couldn't parse: '%.120s'", remote->
buffer + header->payload_offset);
644 struct pollfd fds = { 0, };
648 int timeout = total_timeout;
650 #ifdef HAVE_GNUTLS_GNUTLS_H 651 if (remote->tls_session) {
652 void *sock_ptr = gnutls_transport_get_ptr(*remote->tls_session);
654 sock = GPOINTER_TO_INT(sock_ptr);
661 crm_err(
"Unsupported connection type");
678 if (errno == EINTR && (timeout > 0)) {
679 timeout = total_timeout - ((time(NULL) - start) * 1000);
680 if (timeout < 1000) {
685 rc = poll(&fds, 1, timeout);
686 }
while (rc < 0 && errno == EINTR);
688 return (rc < 0)? -errno : rc;
706 size_t read_len =
sizeof(
struct crm_remote_header_v0);
707 struct crm_remote_header_v0 *header = crm_remote_header(remote);
711 read_len = header->size_total;
717 crm_trace(
"Expanding buffer to %llu bytes",
724 #ifdef HAVE_GNUTLS_GNUTLS_H 725 if (remote->tls_session) {
726 rc = gnutls_record_recv(*(remote->tls_session),
729 if (rc == GNUTLS_E_INTERRUPTED) {
731 }
else if (rc == GNUTLS_E_AGAIN) {
734 crm_debug(
"TLS receive failed: %s (%d)", gnutls_strerror(rc), rc);
750 crm_err(
"Unsupported connection type");
751 return -ESOCKTNOSUPPORT;
759 crm_trace(
"Received %u more bytes, %llu total",
762 }
else if (rc == -EINTR || rc == -EAGAIN) {
765 }
else if (rc == 0) {
766 crm_debug(
"EOF encoutered after %llu bytes",
771 crm_debug(
"Error receiving message after %llu bytes: %s (%d)",
777 header = crm_remote_header(remote);
780 crm_trace(
"Read less than the advertised length: %llu < %u bytes",
784 crm_trace(
"Read full message of %llu bytes",
807 time_t start = time(NULL);
808 int remaining_timeout = 0;
810 if (total_timeout == 0) {
811 total_timeout = 10000;
812 }
else if (total_timeout < 0) {
813 total_timeout = 60000;
817 remaining_timeout = total_timeout;
818 while ((remaining_timeout > 0) && !(*disconnected)) {
820 crm_trace(
"Waiting for remote data (%d of %d ms timeout remaining)",
821 remaining_timeout, total_timeout);
825 crm_err(
"Timed out (%d ms) while waiting for remote data",
830 crm_debug(
"Wait for remote data aborted, will try again: %s " 834 rc = crm_remote_recv_once(remote);
837 }
else if (rc == -EAGAIN) {
838 crm_trace(
"Still waiting for remote data");
845 if (rc == -ENOTCONN) {
850 remaining_timeout = total_timeout - ((time(NULL) - start) * 1000);
856 struct tcp_async_cb_data {
860 void (*callback) (
void *userdata,
int sock);
866 check_connect_finished(gpointer userdata)
868 struct tcp_async_cb_data *cb_data = userdata;
870 int sock = cb_data->sock;
874 socklen_t len =
sizeof(error);
875 struct timeval ts = { 0, };
877 if (cb_data->success == TRUE) {
885 crm_trace(
"fd %d: checking to see if connect finished", sock);
886 cb_arg = select(sock + 1, &rset, &wset, NULL, &ts);
890 if ((errno == EINPROGRESS) || (errno == EAGAIN)) {
892 if ((time(NULL) - cb_data->start) < (cb_data->timeout / 1000)) {
898 crm_trace(
"fd %d: select failed %d connect dispatch ", sock, cb_arg);
900 }
else if (cb_arg == 0) {
901 if ((time(NULL) - cb_data->start) < (cb_data->timeout / 1000)) {
904 crm_debug(
"fd %d: timeout during select", sock);
908 crm_trace(
"fd %d: select returned success", sock);
913 if (FD_ISSET(sock, &rset) || FD_ISSET(sock, &wset)) {
914 if (getsockopt(sock, SOL_SOCKET, SO_ERROR, &error, &len) < 0) {
916 crm_trace(
"fd %d: call to getsockopt failed", sock);
920 crm_trace(
"fd %d: error returned from getsockopt: %d", sock, error);
925 crm_trace(
"neither read nor write set after select");
934 cb_arg = cb_data->sock;
940 if (cb_data->callback) {
941 cb_data->callback(cb_data->userdata, cb_arg);
953 internal_tcp_connect_async(
int sock,
954 const struct sockaddr *addr, socklen_t addrlen,
int timeout ,
955 int *timer_id,
void *userdata,
void (*callback) (
void *userdata,
int sock))
960 struct tcp_async_cb_data *cb_data = NULL;
964 crm_warn(
"Could not set socket non-blocking: %s " CRM_XS " rc=%d",
970 rc = connect(sock, addr, addrlen);
971 if (rc < 0 && (errno != EINPROGRESS) && (errno != EAGAIN)) {
976 cb_data = calloc(1,
sizeof(
struct tcp_async_cb_data));
977 cb_data->userdata = userdata;
978 cb_data->callback = callback;
979 cb_data->sock = sock;
980 cb_data->timeout = timeout;
981 cb_data->start = time(NULL);
988 cb_data->success = TRUE;
1000 crm_trace(
"Scheduling check in %dms for whether connect to fd %d finished",
1002 timer = g_timeout_add(interval, check_connect_finished, cb_data);
1011 internal_tcp_connect(
int sock,
const struct sockaddr *addr, socklen_t addrlen)
1013 int rc = connect(sock, addr, addrlen);
1024 crm_warn(
"Could not set socket non-blocking: %s " CRM_XS " rc=%d",
1047 int *timer_id,
void *userdata,
1048 void (*callback) (
void *userdata,
int sock))
1050 char buffer[INET6_ADDRSTRLEN];
1051 struct addrinfo *res = NULL;
1052 struct addrinfo *rp = NULL;
1053 struct addrinfo hints;
1054 const char *server =
host;
1056 int sock = -ENOTCONN;
1059 memset(&hints, 0,
sizeof(
struct addrinfo));
1060 hints.ai_family = AF_UNSPEC;
1061 hints.ai_socktype = SOCK_STREAM;
1062 hints.ai_flags = AI_CANONNAME;
1063 ret_ga = getaddrinfo(server, NULL, &hints, &res);
1065 crm_err(
"Unable to get IP address info for %s: %s",
1066 server, gai_strerror(ret_ga));
1069 if (!res || !res->ai_addr) {
1070 crm_err(
"Unable to get IP address info for %s: no result", server);
1075 for (rp = res; rp != NULL; rp = rp->ai_next) {
1076 struct sockaddr *addr = rp->ai_addr;
1082 if (rp->ai_canonname) {
1083 server = res->ai_canonname;
1085 crm_debug(
"Got canonical name %s for %s", server, host);
1087 sock = socket(rp->ai_family, SOCK_STREAM, IPPROTO_TCP);
1089 crm_perror(LOG_WARNING,
"creating socket for connection to %s",
1097 if (addr->sa_family == AF_INET6) {
1098 ((
struct sockaddr_in6 *)(
void*)addr)->sin6_port = htons(port);
1100 ((
struct sockaddr_in *)(
void*)addr)->sin_port = htons(port);
1103 memset(buffer, 0,
DIMOF(buffer));
1105 crm_info(
"Attempting TCP connection to %s:%d", buffer, port);
1108 if (internal_tcp_connect_async
1109 (sock, rp->ai_addr, rp->ai_addrlen, timeout, timer_id, userdata, callback) == 0) {
1113 }
else if (internal_tcp_connect(sock, rp->ai_addr, rp->ai_addrlen) == 0) {
1148 switch (((
struct sockaddr*)sa)->sa_family) {
1150 inet_ntop(AF_INET, &(((
struct sockaddr_in *)sa)->sin_addr),
1151 s, INET6_ADDRSTRLEN);
1155 inet_ntop(AF_INET6, &(((
struct sockaddr_in6 *)sa)->sin6_addr),
1156 s, INET6_ADDRSTRLEN);
1160 strcpy(s,
"<invalid>");
1170 struct sockaddr_storage addr;
1171 char addr_str[INET6_ADDRSTRLEN];
1172 #ifdef TCP_USER_TIMEOUT 1178 laddr =
sizeof(addr);
1179 memset(&addr, 0,
sizeof(addr));
1180 csock = accept(ssock, (
struct sockaddr *)&addr, &laddr);
1182 crm_info(
"New remote connection from %s", addr_str);
1185 crm_err(
"accept socket failed");
1191 crm_err(
"Could not set socket non-blocking: %s " CRM_XS " rc=%d",
1197 #ifdef TCP_USER_TIMEOUT 1198 if (sbd_timeout > 0) {
1199 optval = sbd_timeout / 2;
1200 rc = setsockopt(csock, SOL_TCP, TCP_USER_TIMEOUT,
1201 &optval,
sizeof(optval));
1203 crm_err(
"setting TCP_USER_TIMEOUT (%d) on client socket failed",
1222 static int port = 0;
1225 const char *env = getenv(
"PCMK_remote_port");
1229 port = strtol(env, NULL, 10);
1230 if (errno || (port < 1) || (port > 65535)) {
1231 crm_warn(
"Environment variable PCMK_remote_port has invalid value '%s', using %d instead",
long crm_get_sbd_timeout(void)
uint32_t payload_compressed
const char * pcmk_strerror(int rc)
int crm_parse_int(const char *text, const char *default_text)
Parse an integer value from a string.
struct crm_remote_s * remote
uint32_t payload_uncompressed
#define CRM_LOG_ASSERT(expr)
int crm_default_remote_port()
Get the default remote connection TCP port on this host.
Wrappers for and extensions to glib mainloop.
xmlNode * string2xml(const char *input)
int crm_remote_send(crm_remote_t *remote, xmlNode *msg)
#define DEFAULT_REMOTE_PORT
#define PCMK_GNUTLS_PRIORITIES
int crm_remote_tcp_connect(const char *host, int port)
#define crm_warn(fmt, args...)
#define crm_debug(fmt, args...)
void crm_sockaddr2str(void *sa, char *s)
Convert an IP address (IPv4 or IPv6) to a string for logging.
int crm_remote_accept(int ssock)
#define crm_trace(fmt, args...)
int crm_set_nonblocking(int fd)
int crm_remote_ready(crm_remote_t *remote, int total_timeout)
Wrappers for and extensions to libxml2.
struct tcp_async_cb_data __attribute__
#define crm_perror(level, fmt, args...)
Log a system error message.
#define REMOTE_MSG_VERSION
#define crm_err(fmt, args...)
const char * bz2_strerror(int rc)
char * dump_xml_unformatted(xmlNode *msg)
xmlNode * crm_remote_parse_buffer(crm_remote_t *remote)
char * crm_strdup_printf(char const *format,...) __attribute__((__format__(__printf__
int crm_remote_tcp_connect_async(const char *host, int port, int timeout, int *timer_id, void *userdata, void(*callback)(void *userdata, int sock))
#define crm_info(fmt, args...)
gboolean crm_remote_recv(crm_remote_t *remote, int total_timeout, int *disconnected)