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>
44 #ifdef HAVE_GNUTLS_GNUTLS_H
46 # include <gnutls/gnutls.h>
48 const int psk_tls_kx_order[] = {
53 const int anon_tls_kx_order[] = {
63 #ifdef HAVE_LINUX_SWAB_H
64 # include <linux/swab.h>
70 #define __swab16(x) ((uint16_t)( \
71 (((uint16_t)(x) & (uint16_t)0x00ffU) << 8) | \
72 (((uint16_t)(x) & (uint16_t)0xff00U) >> 8)))
74 #define __swab32(x) ((uint32_t)( \
75 (((uint32_t)(x) & (uint32_t)0x000000ffUL) << 24) | \
76 (((uint32_t)(x) & (uint32_t)0x0000ff00UL) << 8) | \
77 (((uint32_t)(x) & (uint32_t)0x00ff0000UL) >> 8) | \
78 (((uint32_t)(x) & (uint32_t)0xff000000UL) >> 24)))
80 #define __swab64(x) ((uint64_t)( \
81 (((uint64_t)(x) & (uint64_t)0x00000000000000ffULL) << 56) | \
82 (((uint64_t)(x) & (uint64_t)0x000000000000ff00ULL) << 40) | \
83 (((uint64_t)(x) & (uint64_t)0x0000000000ff0000ULL) << 24) | \
84 (((uint64_t)(x) & (uint64_t)0x00000000ff000000ULL) << 8) | \
85 (((uint64_t)(x) & (uint64_t)0x000000ff00000000ULL) >> 8) | \
86 (((uint64_t)(x) & (uint64_t)0x0000ff0000000000ULL) >> 24) | \
87 (((uint64_t)(x) & (uint64_t)0x00ff000000000000ULL) >> 40) | \
88 (((uint64_t)(x) & (uint64_t)0xff00000000000000ULL) >> 56)))
91 #define REMOTE_MSG_VERSION 1
92 #define ENDIAN_LOCAL 0xBADADBBD
94 struct crm_remote_header_v0
109 static struct crm_remote_header_v0 *
112 struct crm_remote_header_v0 *header = (
struct crm_remote_header_v0 *)remote->
buffer;
113 if(remote->
buffer_offset <
sizeof(
struct crm_remote_header_v0)) {
121 crm_err(
"Invalid message detected, endian mismatch: %lx is neither %lx nor the swab'd %lx",
127 header->flags =
__swab64(header->flags);
128 header->endian =
__swab32(header->endian);
130 header->version =
__swab32(header->version);
131 header->size_total =
__swab32(header->size_total);
132 header->payload_offset =
__swab32(header->payload_offset);
133 header->payload_compressed =
__swab32(header->payload_compressed);
134 header->payload_uncompressed =
__swab32(header->payload_uncompressed);
140 #ifdef HAVE_GNUTLS_GNUTLS_H
147 time_t start = time(NULL);
150 rc = gnutls_handshake(*remote->tls_session);
151 if (rc == GNUTLS_E_INTERRUPTED || rc == GNUTLS_E_AGAIN) {
159 }
while (((time(NULL) - start) < (timeout_ms / 1000)) &&
160 (rc == GNUTLS_E_INTERRUPTED || rc == GNUTLS_E_AGAIN));
163 crm_trace(
"gnutls_handshake() failed with %d", rc);
174 gnutls_init(session, type);
175 # ifdef HAVE_GNUTLS_PRIORITY_SET_DIRECT
177 gnutls_priority_set_direct(*session,
"NORMAL:+ANON-DH", NULL);
180 gnutls_set_default_priority(*session);
181 gnutls_kx_set_priority(*session, anon_tls_kx_order);
183 gnutls_transport_set_ptr(*session, (gnutls_transport_ptr_t) GINT_TO_POINTER(csock));
186 gnutls_credentials_set(*session, GNUTLS_CRD_ANON,
187 (gnutls_anon_server_credentials_t) credentials);
190 gnutls_credentials_set(*session, GNUTLS_CRD_ANON,
191 (gnutls_anon_client_credentials_t) credentials);
203 gnutls_init(session, type);
204 # ifdef HAVE_GNUTLS_PRIORITY_SET_DIRECT
205 gnutls_priority_set_direct(*session,
"NORMAL:+DHE-PSK:+PSK", NULL);
207 gnutls_set_default_priority(*session);
208 gnutls_kx_set_priority(*session, psk_tls_kx_order);
210 gnutls_transport_set_ptr(*session, (gnutls_transport_ptr_t) GINT_TO_POINTER(csock));
213 gnutls_credentials_set(*session, GNUTLS_CRD_PSK,
214 (gnutls_psk_server_credentials_t) credentials);
217 gnutls_credentials_set(*session, GNUTLS_CRD_PSK,
218 (gnutls_psk_client_credentials_t) credentials);
228 const char *unsent = buf;
237 crm_trace(
"Message size: %llu", (
unsigned long long) len);
240 rc = gnutls_record_send(*session, unsent, len);
242 if (rc == GNUTLS_E_INTERRUPTED || rc == GNUTLS_E_AGAIN) {
246 crm_err(
"Connection terminated rc = %d", rc);
249 }
else if (rc < len) {
250 crm_debug(
"Sent %d of %llu bytes", rc, (
unsigned long long) len);
259 return rc < 0 ? rc : total_send;
264 crm_send_plaintext(
int sock,
const char *buf,
size_t len)
268 const char *unsent = buf;
276 crm_trace(
"Message on socket %d: size=%llu",
277 sock, (
unsigned long long) len);
279 rc = write(sock, unsent, len);
287 crm_perror(LOG_ERR,
"Could only write %d of the remaining %d bytes", rc, (
int)len);
291 }
else if (rc < len) {
292 crm_trace(
"Only sent %d of %llu remaining bytes",
293 rc, (
unsigned long long) len);
299 crm_trace(
"Sent %d bytes: %.100s", rc, buf);
302 return rc < 0 ? rc : total_send;
307 crm_remote_sendv(
crm_remote_t * remote,
struct iovec * iov,
int iovs)
310 int rc = -ESOCKTNOSUPPORT;
312 for(; lpc < iovs; lpc++) {
314 #ifdef HAVE_GNUTLS_GNUTLS_H
315 if (remote->tls_session) {
316 rc = crm_send_tls(remote->tls_session, iov[lpc].iov_base, iov[lpc].iov_len);
321 rc = crm_send_plaintext(remote->
tcp_socket, iov[lpc].iov_base, iov[lpc].iov_len);
324 crm_err(
"Unsupported connection type");
334 static uint64_t
id = 0;
338 struct crm_remote_header_v0 *header;
340 if (xml_text == NULL) {
341 crm_err(
"Invalid XML, can not send msg");
345 header = calloc(1,
sizeof(
struct crm_remote_header_v0));
346 iov[0].iov_base = header;
347 iov[0].iov_len =
sizeof(
struct crm_remote_header_v0);
349 iov[1].iov_base = xml_text;
350 iov[1].iov_len = 1 + strlen(xml_text);
356 header->payload_offset = iov[0].iov_len;
357 header->payload_uncompressed = iov[1].iov_len;
358 header->size_total = iov[0].iov_len + iov[1].iov_len;
360 crm_trace(
"Sending len[0]=%d, start=%x\n",
361 (
int)iov[0].iov_len, *(
int*)(
void*)xml_text);
362 rc = crm_remote_sendv(remote, iov, 2);
364 crm_err(
"Failed to send remote msg, rc = %d", rc);
367 free(iov[0].iov_base);
368 free(iov[1].iov_base);
382 struct crm_remote_header_v0 *header = crm_remote_header(remote);
384 if (remote->
buffer == NULL || header == NULL) {
389 if (header->payload_compressed) {
391 unsigned int size_u = 1 + header->payload_uncompressed;
392 char *uncompressed = calloc(1, header->payload_offset + size_u);
394 crm_trace(
"Decompressing message data %d bytes into %d bytes",
395 header->payload_compressed, size_u);
397 rc = BZ2_bzBuffToBuffDecompress(uncompressed + header->payload_offset, &size_u,
398 remote->
buffer + header->payload_offset,
399 header->payload_compressed, 1, 0);
402 crm_warn(
"Couldn't decompress v%d message, we only understand v%d",
407 }
else if (rc != BZ_OK) {
413 CRM_ASSERT(size_u == header->payload_uncompressed);
415 memcpy(uncompressed, remote->
buffer, header->payload_offset);
416 remote->
buffer_size = header->payload_offset + size_u;
419 remote->
buffer = uncompressed;
420 header = crm_remote_header(remote);
426 CRM_LOG_ASSERT(remote->
buffer[
sizeof(
struct crm_remote_header_v0) + header->payload_uncompressed - 1] == 0);
430 crm_warn(
"Couldn't parse v%d message, we only understand v%d",
433 }
else if (xml == NULL) {
434 crm_err(
"Couldn't parse: '%.120s'", remote->
buffer + header->payload_offset);
451 struct pollfd fds = { 0, };
456 #ifdef HAVE_GNUTLS_GNUTLS_H
457 if (remote->tls_session) {
458 void *sock_ptr = gnutls_transport_get_ptr(*remote->tls_session);
460 sock = GPOINTER_TO_INT(sock_ptr);
467 crm_err(
"Unsupported connection type");
484 if (errno == EINTR && (timeout > 0)) {
485 timeout = timeout - ((time(NULL) - start) * 1000);
486 if (timeout < 1000) {
491 rc = poll(&fds, 1, timeout);
492 }
while (rc < 0 && errno == EINTR);
512 size_t read_len =
sizeof(
struct crm_remote_header_v0);
513 struct crm_remote_header_v0 *header = crm_remote_header(remote);
517 read_len = header->size_total;
523 crm_trace(
"Expanding buffer to %llu bytes",
530 #ifdef HAVE_GNUTLS_GNUTLS_H
531 if (remote->tls_session) {
532 rc = gnutls_record_recv(*(remote->tls_session),
535 if (rc == GNUTLS_E_INTERRUPTED) {
537 }
else if (rc == GNUTLS_E_AGAIN) {
540 crm_debug(
"TLS receive failed: %s (%d)", gnutls_strerror(rc), rc);
556 crm_err(
"Unsupported connection type");
557 return -ESOCKTNOSUPPORT;
565 crm_trace(
"Received %u more bytes, %llu total",
568 }
else if (rc == -EINTR || rc == -EAGAIN) {
571 }
else if (rc == 0) {
572 crm_debug(
"EOF encoutered after %llu bytes",
577 crm_debug(
"Error receiving message after %llu bytes: %s (%d)",
583 header = crm_remote_header(remote);
586 crm_trace(
"Read less than the advertised length: %llu < %u bytes",
590 crm_trace(
"Read full message of %llu bytes",
610 time_t start = time(NULL);
611 int remaining_timeout = 0;
613 if (total_timeout == 0) {
614 total_timeout = 10000;
615 }
else if (total_timeout < 0) {
616 total_timeout = 60000;
620 remaining_timeout = total_timeout;
621 while ((remaining_timeout > 0) && !(*disconnected)) {
624 crm_trace(
"waiting to receive remote msg, starting timeout %d, remaining_timeout %d",
625 total_timeout, remaining_timeout);
629 crm_err(
"poll timed out (%d ms) while waiting to receive msg", remaining_timeout);
632 }
else if (rc == -EAGAIN) {
633 crm_trace(
"waiting for remote connection data (up to %dms)",
640 rc = crm_remote_recv_once(remote);
648 if(rc == -ENOTCONN) {
653 remaining_timeout = remaining_timeout - ((time(NULL) - start) * 1000);
659 struct tcp_async_cb_data {
663 void (*callback) (
void *userdata,
int sock);
669 check_connect_finished(gpointer userdata)
671 struct tcp_async_cb_data *cb_data = userdata;
673 int sock = cb_data->sock;
677 socklen_t len =
sizeof(error);
678 struct timeval ts = { 0, };
680 if (cb_data->success == TRUE) {
688 crm_trace(
"fd %d: checking to see if connect finished", sock);
689 rc = select(sock + 1, &rset, &wset, NULL, &ts);
693 if ((errno == EINPROGRESS) || (errno == EAGAIN)) {
695 if ((time(NULL) - cb_data->start) < (cb_data->timeout / 1000)) {
701 crm_trace(
"fd %d: select failed %d connect dispatch ", sock, rc);
703 }
else if (rc == 0) {
704 if ((time(NULL) - cb_data->start) < (cb_data->timeout / 1000)) {
707 crm_debug(
"fd %d: timeout during select", sock);
711 crm_trace(
"fd %d: select returned success", sock);
716 if (FD_ISSET(sock, &rset) || FD_ISSET(sock, &wset)) {
717 if (getsockopt(sock, SOL_SOCKET, SO_ERROR, &error, &len) < 0) {
718 crm_trace(
"fd %d: call to getsockopt failed", sock);
724 crm_trace(
"fd %d: error returned from getsockopt: %d", sock, error);
729 crm_trace(
"neither read nor write set after select");
744 if (cb_data->callback) {
745 cb_data->callback(cb_data->userdata, rc);
757 internal_tcp_connect_async(
int sock,
758 const struct sockaddr *addr, socklen_t addrlen,
int timeout ,
759 int *timer_id,
void *userdata,
void (*callback) (
void *userdata,
int sock))
765 struct tcp_async_cb_data *cb_data = NULL;
767 if ((flag = fcntl(sock, F_GETFL)) >= 0) {
768 if (fcntl(sock, F_SETFL, flag | O_NONBLOCK) < 0) {
769 crm_err(
"fcntl() write failed");
774 rc = connect(sock, addr, addrlen);
776 if (rc < 0 && (errno != EINPROGRESS) && (errno != EAGAIN)) {
780 cb_data = calloc(1,
sizeof(
struct tcp_async_cb_data));
781 cb_data->userdata = userdata;
782 cb_data->callback = callback;
783 cb_data->sock = sock;
784 cb_data->timeout = timeout;
785 cb_data->start = time(NULL);
792 cb_data->success = TRUE;
804 crm_trace(
"fd %d: scheduling to check if connect finished in %dms second", sock, interval);
805 timer = g_timeout_add(interval, check_connect_finished, cb_data);
814 internal_tcp_connect(
int sock,
const struct sockaddr *addr, socklen_t addrlen)
817 int rc = connect(sock, addr, addrlen);
820 if ((flag = fcntl(sock, F_GETFL)) >= 0) {
821 if (fcntl(sock, F_SETFL, flag | O_NONBLOCK) < 0) {
822 crm_err(
"fcntl() write failed");
839 int *timer_id,
void *userdata,
void (*callback) (
void *userdata,
int sock))
842 struct addrinfo *res = NULL;
843 struct addrinfo *rp = NULL;
844 struct addrinfo hints;
845 const char *server =
host;
850 memset(&hints, 0,
sizeof(
struct addrinfo));
851 hints.ai_family = AF_UNSPEC;
852 hints.ai_socktype = SOCK_STREAM;
853 hints.ai_flags = AI_CANONNAME;
856 ret_ga = getaddrinfo(server, NULL, &hints, &res);
858 crm_err(
"getaddrinfo: %s", gai_strerror(ret_ga));
862 if (!res || !res->ai_addr) {
867 for (rp = res; rp != NULL; rp = rp->ai_next) {
868 struct sockaddr *addr = rp->ai_addr;
874 if (rp->ai_canonname) {
875 server = res->ai_canonname;
877 crm_debug(
"Got address %s for %s", server, host);
880 sock = socket(rp->ai_family, SOCK_STREAM, IPPROTO_TCP);
882 crm_err(
"Socket creation failed for remote client connection.");
886 memset(buffer, 0,
DIMOF(buffer));
887 if (addr->sa_family == AF_INET6) {
888 struct sockaddr_in6 *addr_in = (
struct sockaddr_in6 *)(
void*)addr;
890 addr_in->sin6_port = htons(port);
891 inet_ntop(addr->sa_family, &addr_in->sin6_addr, buffer,
DIMOF(buffer));
894 struct sockaddr_in *addr_in = (
struct sockaddr_in *)(
void*)addr;
896 addr_in->sin_port = htons(port);
897 inet_ntop(addr->sa_family, &addr_in->sin_addr, buffer,
DIMOF(buffer));
900 crm_info(
"Attempting to connect to remote server at %s:%d", buffer, port);
903 if (internal_tcp_connect_async
904 (sock, rp->ai_addr, rp->ai_addrlen, timeout, timer_id, userdata, callback) == 0) {
909 if (internal_tcp_connect(sock, rp->ai_addr, rp->ai_addrlen) == 0) {
uint32_t payload_compressed
const char * pcmk_strerror(int rc)
int crm_remote_ready(crm_remote_t *remote, int timeout)
uint32_t payload_uncompressed
#define CRM_LOG_ASSERT(expr)
Wrappers for and extensions to glib mainloop.
xmlNode * string2xml(const char *input)
int crm_remote_send(crm_remote_t *remote, xmlNode *msg)
int crm_remote_tcp_connect(const char *host, int port)
#define crm_warn(fmt, args...)
#define crm_debug(fmt, args...)
int crm_initiate_client_tls_handshake(crm_remote_t *remote, int timeout_ms)
#define crm_trace(fmt, args...)
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)
void * create_psk_tls_session(int csock, int type, void *credentials)
void * crm_create_anon_tls_session(int sock, int type, void *credentials)
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)
enum crm_ais_msg_types type