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 143 crm_initiate_client_tls_handshake(
crm_remote_t * remote,
int timeout_ms)
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);
169 crm_create_anon_tls_session(
int csock,
int type ,
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);
199 create_psk_tls_session(
int csock,
int type ,
void *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;
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 %d bytes", rc, 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=%d", sock, len);
278 rc = write(sock, unsent, len);
286 crm_perror(LOG_ERR,
"Could only write %d of the remaining %d bytes", rc, (
int)len);
290 }
else if (rc < len) {
291 crm_trace(
"Only sent %d of %d remaining bytes", rc, len);
297 crm_trace(
"Sent %d bytes: %.100s", rc, buf);
300 return rc < 0 ? rc : total_send;
305 crm_remote_sendv(
crm_remote_t * remote,
struct iovec * iov,
int iovs)
308 int rc = -ESOCKTNOSUPPORT;
310 for(; lpc < iovs; lpc++) {
312 #ifdef HAVE_GNUTLS_GNUTLS_H 313 if (remote->tls_session) {
314 rc = crm_send_tls(remote->tls_session, iov[lpc].iov_base, iov[lpc].iov_len);
319 rc = crm_send_plaintext(remote->
tcp_socket, iov[lpc].iov_base, iov[lpc].iov_len);
322 crm_err(
"Unsupported connection type");
332 static uint64_t
id = 0;
336 struct crm_remote_header_v0 *header;
338 if (xml_text == NULL) {
339 crm_err(
"Invalid XML, can not send msg");
343 header = calloc(1,
sizeof(
struct crm_remote_header_v0));
344 iov[0].iov_base = header;
345 iov[0].iov_len =
sizeof(
struct crm_remote_header_v0);
347 iov[1].iov_base = xml_text;
348 iov[1].iov_len = 1 + strlen(xml_text);
354 header->payload_offset = iov[0].iov_len;
355 header->payload_uncompressed = iov[1].iov_len;
356 header->size_total = iov[0].iov_len + iov[1].iov_len;
358 crm_trace(
"Sending len[0]=%d, start=%x\n",
359 (
int)iov[0].iov_len, *(
int*)(
void*)xml_text);
360 rc = crm_remote_sendv(remote, iov, 2);
362 crm_err(
"Failed to send remote msg, rc = %d", rc);
365 free(iov[0].iov_base);
366 free(iov[1].iov_base);
380 struct crm_remote_header_v0 *header = crm_remote_header(remote);
382 if (remote->
buffer == NULL || header == NULL) {
390 if (header->payload_compressed) {
392 unsigned int size_u = 1 + header->payload_uncompressed;
393 char *uncompressed = calloc(1, header->payload_offset + size_u);
395 crm_trace(
"Decompressing message data %d bytes into %d bytes",
396 header->payload_compressed, size_u);
398 rc = BZ2_bzBuffToBuffDecompress(uncompressed + header->payload_offset, &size_u,
399 remote->
buffer + header->payload_offset,
400 header->payload_compressed, 1, 0);
403 crm_warn(
"Couldn't decompress v%d message, we only understand v%d",
408 }
else if (rc != BZ_OK) {
414 CRM_ASSERT(size_u == header->payload_uncompressed);
416 memcpy(uncompressed, remote->
buffer, header->payload_offset);
417 remote->
buffer_size = header->payload_offset + size_u;
420 remote->
buffer = uncompressed;
421 header = crm_remote_header(remote);
424 CRM_LOG_ASSERT(remote->
buffer[
sizeof(
struct crm_remote_header_v0) + header->payload_uncompressed - 1] == 0);
428 crm_warn(
"Couldn't parse v%d message, we only understand v%d",
431 }
else if (xml == NULL) {
432 crm_err(
"Couldn't parse: '%.120s'", remote->
buffer + header->payload_offset);
449 struct pollfd fds = { 0, };
454 #ifdef HAVE_GNUTLS_GNUTLS_H 455 if (remote->tls_session) {
456 void *sock_ptr = gnutls_transport_get_ptr(*remote->tls_session);
458 sock = GPOINTER_TO_INT(sock_ptr);
465 crm_err(
"Unsupported connection type");
482 if (errno == EINTR && (timeout > 0)) {
483 timeout = timeout - ((time(NULL) - start) * 1000);
484 if (timeout < 1000) {
489 rc = poll(&fds, 1, timeout);
490 }
while (rc < 0 && errno == EINTR);
510 size_t read_len =
sizeof(
struct crm_remote_header_v0);
511 struct crm_remote_header_v0 *header = crm_remote_header(remote);
515 read_len = header->size_total;
527 #ifdef HAVE_GNUTLS_GNUTLS_H 528 if (remote->tls_session) {
529 rc = gnutls_record_recv(*(remote->tls_session),
532 if (rc == GNUTLS_E_INTERRUPTED) {
534 }
else if (rc == GNUTLS_E_AGAIN) {
537 crm_debug(
"TLS receive failed: %s (%d)", gnutls_strerror(rc), rc);
553 crm_err(
"Unsupported connection type");
554 return -ESOCKTNOSUPPORT;
564 }
else if (rc == -EINTR || rc == -EAGAIN) {
567 }
else if (rc == 0) {
572 crm_debug(
"Error receiving message after %u bytes: %s (%d)",
577 header = crm_remote_header(remote);
580 crm_trace(
"Read less than the advertised length: %u < %u bytes",
602 time_t start = time(NULL);
603 int remaining_timeout = 0;
605 if (total_timeout == 0) {
606 total_timeout = 10000;
607 }
else if (total_timeout < 0) {
608 total_timeout = 60000;
612 remaining_timeout = total_timeout;
613 while ((remaining_timeout > 0) && !(*disconnected)) {
616 crm_trace(
"waiting to receive remote msg, starting timeout %d, remaining_timeout %d",
617 total_timeout, remaining_timeout);
621 crm_err(
"poll timed out (%d ms) while waiting to receive msg", remaining_timeout);
624 }
else if (rc == -EAGAIN) {
625 crm_trace(
"waiting for remote connection data (up to %dms)",
632 rc = crm_remote_recv_once(remote);
640 if(rc == -ENOTCONN) {
645 remaining_timeout = remaining_timeout - ((time(NULL) - start) * 1000);
651 struct tcp_async_cb_data {
655 void (*callback) (
void *userdata,
int sock);
661 check_connect_finished(gpointer userdata)
663 struct tcp_async_cb_data *cb_data = userdata;
665 int sock = cb_data->sock;
669 socklen_t len =
sizeof(error);
670 struct timeval ts = { 0, };
672 if (cb_data->success == TRUE) {
680 crm_trace(
"fd %d: checking to see if connect finished", sock);
681 rc = select(sock + 1, &rset, &wset, NULL, &ts);
685 if ((errno == EINPROGRESS) || (errno == EAGAIN)) {
687 if ((time(NULL) - cb_data->start) < (cb_data->timeout / 1000)) {
693 crm_trace(
"fd %d: select failed %d connect dispatch ", rc);
695 }
else if (rc == 0) {
696 if ((time(NULL) - cb_data->start) < (cb_data->timeout / 1000)) {
699 crm_debug(
"fd %d: timeout during select", sock);
703 crm_trace(
"fd %d: select returned success", sock);
708 if (FD_ISSET(sock, &rset) || FD_ISSET(sock, &wset)) {
709 if (getsockopt(sock, SOL_SOCKET, SO_ERROR, &error, &len) < 0) {
710 crm_trace(
"fd %d: call to getsockopt failed", sock);
716 crm_trace(
"fd %d: error returned from getsockopt: %d", sock, error);
721 crm_trace(
"neither read nor write set after select");
736 if (cb_data->callback) {
737 cb_data->callback(cb_data->userdata, rc);
749 internal_tcp_connect_async(
int sock,
750 const struct sockaddr *addr, socklen_t addrlen,
int timeout ,
751 int *timer_id,
void *userdata,
void (*callback) (
void *userdata,
int sock))
757 struct tcp_async_cb_data *cb_data = NULL;
759 if ((flag = fcntl(sock, F_GETFL)) >= 0) {
760 if (fcntl(sock, F_SETFL, flag | O_NONBLOCK) < 0) {
761 crm_err(
"fcntl() write failed");
766 rc = connect(sock, addr, addrlen);
768 if (rc < 0 && (errno != EINPROGRESS) && (errno != EAGAIN)) {
772 cb_data = calloc(1,
sizeof(
struct tcp_async_cb_data));
773 cb_data->userdata = userdata;
774 cb_data->callback = callback;
775 cb_data->sock = sock;
776 cb_data->timeout = timeout;
777 cb_data->start = time(NULL);
784 cb_data->success = TRUE;
796 crm_trace(
"fd %d: scheduling to check if connect finished in %dms second", sock, interval);
797 timer = g_timeout_add(interval, check_connect_finished, cb_data);
806 internal_tcp_connect(
int sock,
const struct sockaddr *addr, socklen_t addrlen)
809 int rc = connect(sock, addr, addrlen);
812 if ((flag = fcntl(sock, F_GETFL)) >= 0) {
813 if (fcntl(sock, F_SETFL, flag | O_NONBLOCK) < 0) {
814 crm_err(
"fcntl() write failed");
831 int *timer_id,
void *userdata,
void (*callback) (
void *userdata,
int sock))
834 struct addrinfo *res = NULL;
835 struct addrinfo *rp = NULL;
836 struct addrinfo hints;
837 const char *server =
host;
842 memset(&hints, 0,
sizeof(
struct addrinfo));
843 hints.ai_family = AF_UNSPEC;
844 hints.ai_socktype = SOCK_STREAM;
845 hints.ai_flags = AI_CANONNAME;
848 ret_ga = getaddrinfo(server, NULL, &hints, &res);
850 crm_err(
"getaddrinfo: %s", gai_strerror(ret_ga));
854 if (!res || !res->ai_addr) {
859 for (rp = res; rp != NULL; rp = rp->ai_next) {
860 struct sockaddr *addr = rp->ai_addr;
866 if (rp->ai_canonname) {
867 server = res->ai_canonname;
869 crm_debug(
"Got address %s for %s", server, host);
872 sock = socket(rp->ai_family, SOCK_STREAM, IPPROTO_TCP);
874 crm_err(
"Socket creation failed for remote client connection.");
878 memset(buffer, 0,
DIMOF(buffer));
879 if (addr->sa_family == AF_INET6) {
880 struct sockaddr_in6 *addr_in = (
struct sockaddr_in6 *)(
void*)addr;
882 addr_in->sin6_port = htons(port);
883 inet_ntop(addr->sa_family, &addr_in->sin6_addr, buffer,
DIMOF(buffer));
886 struct sockaddr_in *addr_in = (
struct sockaddr_in *)(
void*)addr;
888 addr_in->sin_port = htons(port);
889 inet_ntop(addr->sa_family, &addr_in->sin_addr, buffer,
DIMOF(buffer));
892 crm_info(
"Attempting to connect to remote server at %s:%d", buffer, port);
895 if (internal_tcp_connect_async
896 (sock, rp->ai_addr, rp->ai_addrlen, timeout, timer_id, userdata, callback) == 0) {
901 if (internal_tcp_connect(sock, rp->ai_addr, rp->ai_addrlen) == 0) {
#define REMOTE_MSG_VERSION
const char * pcmk_strerror(int rc)
struct tcp_async_cb_data __attribute__
#define CRM_LOG_ASSERT(expr)
gboolean crm_remote_recv(crm_remote_t *remote, int total_timeout, int *disconnected)
Wrappers for and extensions to glib mainloop.
xmlNode * string2xml(const char *input)
#define crm_warn(fmt, args...)
uint32_t payload_compressed
#define crm_debug(fmt, args...)
int crm_remote_tcp_connect(const char *host, int port)
#define crm_trace(fmt, args...)
Wrappers for and extensions to libxml2.
int crm_remote_ready(crm_remote_t *remote, int timeout)
int crm_remote_tcp_connect_async(const char *host, int port, int timeout, int *timer_id, void *userdata, void(*callback)(void *userdata, int sock))
int crm_remote_send(crm_remote_t *remote, xmlNode *msg)
#define crm_perror(level, fmt, args...)
Log a system error message.
#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)
uint32_t payload_uncompressed
#define crm_info(fmt, args...)
enum crm_ais_msg_types type