pacemaker  1.1.18-2b07d5c5a9
Scalable High-Availability cluster resource manager
ipc.c
Go to the documentation of this file.
1 /*
2  * Copyright 2004-2019 the Pacemaker project contributors
3  *
4  * The version control history for this file may have further details.
5  *
6  * This source code is licensed under the GNU Lesser General Public License
7  * version 2.1 or later (LGPLv2.1+) WITHOUT ANY WARRANTY.
8  */
9 
10 #include <crm_internal.h>
11 
12 #if defined(US_AUTH_PEERCRED_UCRED) || defined(US_AUTH_PEERCRED_SOCKPEERCRED)
13 # ifdef US_AUTH_PEERCRED_UCRED
14 # ifndef _GNU_SOURCE
15 # define _GNU_SOURCE
16 # endif
17 # endif
18 # include <sys/socket.h>
19 #elif defined(US_AUTH_GETPEERUCRED)
20 # include <ucred.h>
21 #endif
22 
23 #include <sys/param.h>
24 
25 #include <stdio.h>
26 #include <sys/types.h>
27 #include <sys/stat.h>
28 #include <unistd.h>
29 #include <grp.h>
30 
31 #include <errno.h>
32 #include <fcntl.h>
33 #include <bzlib.h>
34 
35 #include <crm/crm.h> /* indirectly: pcmk_err_generic */
36 #include <crm/msg_xml.h>
37 #include <crm/common/ipc.h>
38 #include <crm/common/ipcs.h>
39 
40 #include <crm/common/ipc_internal.h> /* PCMK__SPECIAL_PID* */
41 
42 #define PCMK_IPC_VERSION 1
43 
44 /* Evict clients whose event queue grows this large (by default) */
45 #define PCMK_IPC_DEFAULT_QUEUE_MAX 500
46 
47 struct crm_ipc_response_header {
48  struct qb_ipc_response_header qb;
49  uint32_t size_uncompressed;
50  uint32_t size_compressed;
52  uint8_t version; /* Protect against version changes for anyone that might bother to statically link us */
53 };
54 
55 static int hdr_offset = 0;
56 static unsigned int ipc_buffer_max = 0;
57 static unsigned int pick_ipc_buffer(unsigned int max);
58 
59 static inline void
60 crm_ipc_init(void)
61 {
62  if (hdr_offset == 0) {
63  hdr_offset = sizeof(struct crm_ipc_response_header);
64  }
65  if (ipc_buffer_max == 0) {
66  ipc_buffer_max = pick_ipc_buffer(0);
67  }
68 }
69 
70 unsigned int
72 {
73  return pick_ipc_buffer(0);
74 }
75 
76 static char *
77 generateReference(const char *custom1, const char *custom2)
78 {
79  static uint ref_counter = 0;
80  const char *local_cust1 = custom1;
81  const char *local_cust2 = custom2;
82  int reference_len = 4;
83  char *since_epoch = NULL;
84 
85  reference_len += 20; /* too big */
86  reference_len += 40; /* too big */
87 
88  if (local_cust1 == NULL) {
89  local_cust1 = "_empty_";
90  }
91  reference_len += strlen(local_cust1);
92 
93  if (local_cust2 == NULL) {
94  local_cust2 = "_empty_";
95  }
96  reference_len += strlen(local_cust2);
97 
98  since_epoch = calloc(1, reference_len);
99 
100  if (since_epoch != NULL) {
101  sprintf(since_epoch, "%s-%s-%lu-%u",
102  local_cust1, local_cust2, (unsigned long)time(NULL), ref_counter++);
103  }
104 
105  return since_epoch;
106 }
107 
108 xmlNode *
109 create_request_adv(const char *task, xmlNode * msg_data,
110  const char *host_to, const char *sys_to,
111  const char *sys_from, const char *uuid_from, const char *origin)
112 {
113  char *true_from = NULL;
114  xmlNode *request = NULL;
115  char *reference = generateReference(task, sys_from);
116 
117  if (uuid_from != NULL) {
118  true_from = generate_hash_key(sys_from, uuid_from);
119  } else if (sys_from != NULL) {
120  true_from = strdup(sys_from);
121  } else {
122  crm_err("No sys from specified");
123  }
124 
125  /* host_from will get set for us if necessary by CRMd when routed */
126  request = create_xml_node(NULL, __FUNCTION__);
127  crm_xml_add(request, F_CRM_ORIGIN, origin);
128  crm_xml_add(request, F_TYPE, T_CRM);
131  crm_xml_add(request, F_CRM_REFERENCE, reference);
132  crm_xml_add(request, F_CRM_TASK, task);
133  crm_xml_add(request, F_CRM_SYS_TO, sys_to);
134  crm_xml_add(request, F_CRM_SYS_FROM, true_from);
135 
136  /* HOSTTO will be ignored if it is to the DC anyway. */
137  if (host_to != NULL && strlen(host_to) > 0) {
138  crm_xml_add(request, F_CRM_HOST_TO, host_to);
139  }
140 
141  if (msg_data != NULL) {
142  add_message_xml(request, F_CRM_DATA, msg_data);
143  }
144  free(reference);
145  free(true_from);
146 
147  return request;
148 }
149 
150 /*
151  * This method adds a copy of xml_response_data
152  */
153 xmlNode *
154 create_reply_adv(xmlNode * original_request, xmlNode * xml_response_data, const char *origin)
155 {
156  xmlNode *reply = NULL;
157 
158  const char *host_from = crm_element_value(original_request, F_CRM_HOST_FROM);
159  const char *sys_from = crm_element_value(original_request, F_CRM_SYS_FROM);
160  const char *sys_to = crm_element_value(original_request, F_CRM_SYS_TO);
161  const char *type = crm_element_value(original_request, F_CRM_MSG_TYPE);
162  const char *operation = crm_element_value(original_request, F_CRM_TASK);
163  const char *crm_msg_reference = crm_element_value(original_request, F_CRM_REFERENCE);
164 
165  if (type == NULL) {
166  crm_err("Cannot create new_message, no message type in original message");
167  CRM_ASSERT(type != NULL);
168  return NULL;
169 #if 0
170  } else if (strcasecmp(XML_ATTR_REQUEST, type) != 0) {
171  crm_err("Cannot create new_message, original message was not a request");
172  return NULL;
173 #endif
174  }
175  reply = create_xml_node(NULL, __FUNCTION__);
176  if (reply == NULL) {
177  crm_err("Cannot create new_message, malloc failed");
178  return NULL;
179  }
180 
181  crm_xml_add(reply, F_CRM_ORIGIN, origin);
182  crm_xml_add(reply, F_TYPE, T_CRM);
185  crm_xml_add(reply, F_CRM_REFERENCE, crm_msg_reference);
186  crm_xml_add(reply, F_CRM_TASK, operation);
187 
188  /* since this is a reply, we reverse the from and to */
189  crm_xml_add(reply, F_CRM_SYS_TO, sys_from);
190  crm_xml_add(reply, F_CRM_SYS_FROM, sys_to);
191 
192  /* HOSTTO will be ignored if it is to the DC anyway. */
193  if (host_from != NULL && strlen(host_from) > 0) {
194  crm_xml_add(reply, F_CRM_HOST_TO, host_from);
195  }
196 
197  if (xml_response_data != NULL) {
198  add_message_xml(reply, F_CRM_DATA, xml_response_data);
199  }
200 
201  return reply;
202 }
203 
204 /* Libqb based IPC */
205 
206 /* Server... */
207 
208 GHashTable *client_connections = NULL;
209 
210 crm_client_t *
211 crm_client_get(qb_ipcs_connection_t * c)
212 {
213  if (client_connections) {
214  return g_hash_table_lookup(client_connections, c);
215  }
216 
217  crm_trace("No client found for %p", c);
218  return NULL;
219 }
220 
221 crm_client_t *
222 crm_client_get_by_id(const char *id)
223 {
224  gpointer key;
225  crm_client_t *client;
226  GHashTableIter iter;
227 
228  if (client_connections && id) {
229  g_hash_table_iter_init(&iter, client_connections);
230  while (g_hash_table_iter_next(&iter, &key, (gpointer *) & client)) {
231  if (strcmp(client->id, id) == 0) {
232  return client;
233  }
234  }
235  }
236 
237  crm_trace("No client found with id=%s", id);
238  return NULL;
239 }
240 
241 const char *
243 {
244  if (c == NULL) {
245  return "null";
246  } else if (c->name == NULL && c->id == NULL) {
247  return "unknown";
248  } else if (c->name == NULL) {
249  return c->id;
250  } else {
251  return c->name;
252  }
253 }
254 
255 void
257 {
258  if (client_connections == NULL) {
259  crm_trace("Creating client hash table");
260  client_connections = g_hash_table_new(g_direct_hash, g_direct_equal);
261  }
262 }
263 
264 void
266 {
267  if (client_connections != NULL) {
268  int active = g_hash_table_size(client_connections);
269 
270  if (active) {
271  crm_err("Exiting with %d active connections", active);
272  }
273  g_hash_table_destroy(client_connections); client_connections = NULL;
274  }
275 }
276 
277 void
278 crm_client_disconnect_all(qb_ipcs_service_t *service)
279 {
280  qb_ipcs_connection_t *c = NULL;
281 
282  if (service == NULL) {
283  return;
284  }
285 
286  c = qb_ipcs_connection_first_get(service);
287 
288  while (c != NULL) {
289  qb_ipcs_connection_t *last = c;
290 
291  c = qb_ipcs_connection_next_get(service, last);
292 
293  /* There really shouldn't be anyone connected at this point */
294  crm_notice("Disconnecting client %p, pid=%d...", last, crm_ipcs_client_pid(last));
295  qb_ipcs_disconnect(last);
296  qb_ipcs_connection_unref(last);
297  }
298 }
299 
307 crm_client_t *
309 {
310  crm_client_t *client = calloc(1, sizeof(crm_client_t));
311 
312  CRM_ASSERT(client != NULL);
313  client->id = crm_generate_uuid();
314  g_hash_table_insert(client_connections, (key? key : client->id), client);
315  return client;
316 }
317 
318 crm_client_t *
319 crm_client_new(qb_ipcs_connection_t * c, uid_t uid_client, gid_t gid_client)
320 {
321  static gid_t uid_cluster = 0;
322  static gid_t gid_cluster = 0;
323 
324  crm_client_t *client = NULL;
325 
326  CRM_LOG_ASSERT(c);
327  if (c == NULL) {
328  return NULL;
329  }
330 
331  if (uid_cluster == 0) {
332  if (crm_user_lookup(CRM_DAEMON_USER, &uid_cluster, &gid_cluster) < 0) {
333  static bool have_error = FALSE;
334  if(have_error == FALSE) {
335  crm_warn("Could not find user and group IDs for user %s",
337  have_error = TRUE;
338  }
339  }
340  }
341 
342  if (uid_client != 0) {
343  crm_trace("Giving access to group %u", gid_cluster);
344  /* Passing -1 to chown(2) means don't change */
345  qb_ipcs_connection_auth_set(c, -1, gid_cluster, S_IRUSR | S_IWUSR | S_IRGRP | S_IWGRP);
346  }
347 
348  crm_client_init();
349 
350  /* TODO: Do our own auth checking, return NULL if unauthorized */
351  client = crm_client_alloc(c);
352  client->ipcs = c;
353  client->kind = CRM_CLIENT_IPC;
354  client->pid = crm_ipcs_client_pid(c);
355 
356  if ((uid_client == 0) || (uid_client == uid_cluster)) {
357  /* Remember when a connection came from root or hacluster */
359  }
360 
361  crm_debug("Connecting %p for uid=%d gid=%d pid=%u id=%s", c, uid_client, gid_client, client->pid, client->id);
362 
363 #if ENABLE_ACL
364  client->user = uid2username(uid_client);
365 #endif
366  return client;
367 }
368 
369 void
371 {
372  if (c == NULL) {
373  return;
374  }
375 
376  if (client_connections) {
377  if (c->ipcs) {
378  crm_trace("Destroying %p/%p (%d remaining)",
379  c, c->ipcs, crm_hash_table_size(client_connections) - 1);
380  g_hash_table_remove(client_connections, c->ipcs);
381 
382  } else {
383  crm_trace("Destroying remote connection %p (%d remaining)",
384  c, crm_hash_table_size(client_connections) - 1);
385  g_hash_table_remove(client_connections, c->id);
386  }
387  }
388 
389  if (c->event_timer) {
390  g_source_remove(c->event_timer);
391  }
392 
393  crm_debug("Destroying %d events", g_list_length(c->event_queue));
394  while (c->event_queue) {
395  struct iovec *event = c->event_queue->data;
396 
397  c->event_queue = g_list_remove(c->event_queue, event);
398  free(event[0].iov_base);
399  free(event[1].iov_base);
400  free(event);
401  }
402 
403  free(c->id);
404  free(c->name);
405  free(c->user);
406  if (c->remote) {
407  if (c->remote->auth_timeout) {
408  g_source_remove(c->remote->auth_timeout);
409  }
410  free(c->remote->buffer);
411  free(c->remote);
412  }
413  free(c);
414 }
415 
424 bool
425 crm_set_client_queue_max(crm_client_t *client, const char *qmax)
426 {
427  if (is_set(client->flags, crm_client_flag_ipc_privileged)) {
428  int qmax_int = crm_int_helper(qmax, NULL);
429 
430  if ((errno == 0) && (qmax_int > 0)) {
431  client->queue_max = qmax_int;
432  return TRUE;
433  }
434  }
435  return FALSE;
436 }
437 
438 int
439 crm_ipcs_client_pid(qb_ipcs_connection_t * c)
440 {
441  struct qb_ipcs_connection_stats stats;
442 
443  stats.client_pid = 0;
444  qb_ipcs_connection_stats_get(c, &stats, 0);
445  return stats.client_pid;
446 }
447 
448 xmlNode *
450 {
451  xmlNode *xml = NULL;
452  char *uncompressed = NULL;
453  char *text = ((char *)data) + sizeof(struct crm_ipc_response_header);
454  struct crm_ipc_response_header *header = data;
455 
456  if (id) {
457  *id = ((struct qb_ipc_response_header *)data)->id;
458  }
459  if (flags) {
460  *flags = header->flags;
461  }
462 
463  if (is_set(header->flags, crm_ipc_proxied)) {
464  /* mark this client as being the endpoint of a proxy connection.
465  * Proxy connections responses are sent on the event channel to avoid
466  * blocking the proxy daemon (crmd) */
468  }
469 
470  if(header->version > PCMK_IPC_VERSION) {
471  crm_err("Filtering incompatible v%d IPC message, we only support versions <= %d",
472  header->version, PCMK_IPC_VERSION);
473  return NULL;
474  }
475 
476  if (header->size_compressed) {
477  int rc = 0;
478  unsigned int size_u = 1 + header->size_uncompressed;
479  uncompressed = calloc(1, size_u);
480 
481  crm_trace("Decompressing message data %u bytes into %u bytes",
482  header->size_compressed, size_u);
483 
484  rc = BZ2_bzBuffToBuffDecompress(uncompressed, &size_u, text, header->size_compressed, 1, 0);
485  text = uncompressed;
486 
487  if (rc != BZ_OK) {
488  crm_err("Decompression failed: %s (%d)", bz2_strerror(rc), rc);
489  free(uncompressed);
490  return NULL;
491  }
492  }
493 
494  CRM_ASSERT(text[header->size_uncompressed - 1] == 0);
495 
496  crm_trace("Received %.200s", text);
497  xml = string2xml(text);
498 
499  free(uncompressed);
500  return xml;
501 }
502 
504 
505 static gboolean
506 crm_ipcs_flush_events_cb(gpointer data)
507 {
508  crm_client_t *c = data;
509 
510  c->event_timer = 0;
512  return FALSE;
513 }
514 
522 static inline void
523 delay_next_flush(crm_client_t *c, unsigned int queue_len)
524 {
525  /* Delay a maximum of 5 seconds */
526  guint delay = (queue_len < 40)? (1000 + 100 * queue_len) : 5000;
527 
528  c->event_timer = g_timeout_add(delay, crm_ipcs_flush_events_cb, c);
529 }
530 
531 ssize_t
533 {
534  ssize_t rc = 0;
535  unsigned int sent = 0;
536  unsigned int queue_len = 0;
537 
538  if (c == NULL) {
539  return pcmk_ok;
540 
541  } else if (c->event_timer) {
542  /* There is already a timer, wait until it goes off */
543  crm_trace("Timer active for %p - %d", c->ipcs, c->event_timer);
544  return pcmk_ok;
545  }
546 
547  queue_len = g_list_length(c->event_queue);
548  while (c->event_queue && sent < 100) {
549  struct crm_ipc_response_header *header = NULL;
550  struct iovec *event = c->event_queue->data;
551 
552  rc = qb_ipcs_event_sendv(c->ipcs, event, 2);
553  if (rc < 0) {
554  break;
555  }
556 
557  sent++;
558  header = event[0].iov_base;
559  if (header->size_compressed) {
560  crm_trace("Event %d to %p[%d] (%lld compressed bytes) sent",
561  header->qb.id, c->ipcs, c->pid, (long long) rc);
562  } else {
563  crm_trace("Event %d to %p[%d] (%lld bytes) sent: %.120s",
564  header->qb.id, c->ipcs, c->pid, (long long) rc,
565  (char *) (event[1].iov_base));
566  }
567 
568  c->event_queue = g_list_remove(c->event_queue, event);
569  free(event[0].iov_base);
570  free(event[1].iov_base);
571  free(event);
572  }
573 
574  queue_len -= sent;
575  if (sent > 0 || queue_len) {
576  crm_trace("Sent %d events (%d remaining) for %p[%d]: %s (%lld)",
577  sent, queue_len, c->ipcs, c->pid,
578  pcmk_strerror(rc < 0 ? rc : 0), (long long) rc);
579  }
580 
581  if (queue_len) {
582 
583  /* Allow clients to briefly fall behind on processing incoming messages,
584  * but drop completely unresponsive clients so the connection doesn't
585  * consume resources indefinitely.
586  */
587  if (queue_len > QB_MAX(c->queue_max, PCMK_IPC_DEFAULT_QUEUE_MAX)) {
588  if ((c->queue_backlog <= 1) || (queue_len < c->queue_backlog)) {
589  /* Don't evict for a new or shrinking backlog */
590  crm_warn("Client with process ID %u has a backlog of %u messages "
591  CRM_XS " %p", c->pid, queue_len, c->ipcs);
592  } else {
593  crm_err("Evicting client with process ID %u due to backlog of %u messages "
594  CRM_XS " %p", c->pid, queue_len, c->ipcs);
595  c->queue_backlog = 0;
596  qb_ipcs_disconnect(c->ipcs);
597  return rc;
598  }
599  }
600 
601  c->queue_backlog = queue_len;
602  delay_next_flush(c, queue_len);
603 
604  } else {
605  /* Event queue is empty, there is no backlog */
606  c->queue_backlog = 0;
607  }
608 
609  return rc;
610 }
611 
612 ssize_t
613 crm_ipc_prepare(uint32_t request, xmlNode * message, struct iovec ** result, uint32_t max_send_size)
614 {
615  static unsigned int biggest = 0;
616  struct iovec *iov;
617  unsigned int total = 0;
618  char *compressed = NULL;
619  char *buffer = dump_xml_unformatted(message);
620  struct crm_ipc_response_header *header = calloc(1, sizeof(struct crm_ipc_response_header));
621 
622  CRM_ASSERT(result != NULL);
623 
624  crm_ipc_init();
625 
626  if (max_send_size == 0) {
627  max_send_size = ipc_buffer_max;
628  }
629 
630  CRM_LOG_ASSERT(max_send_size != 0);
631 
632  *result = NULL;
633  iov = calloc(2, sizeof(struct iovec));
634 
635 
636  iov[0].iov_len = hdr_offset;
637  iov[0].iov_base = header;
638 
639  header->version = PCMK_IPC_VERSION;
640  header->size_uncompressed = 1 + strlen(buffer);
641  total = iov[0].iov_len + header->size_uncompressed;
642 
643  if (total < max_send_size) {
644  iov[1].iov_base = buffer;
645  iov[1].iov_len = header->size_uncompressed;
646 
647  } else {
648  unsigned int new_size = 0;
649 
651  (buffer, header->size_uncompressed, max_send_size, &compressed, &new_size)) {
652 
653  header->flags |= crm_ipc_compressed;
654  header->size_compressed = new_size;
655 
656  iov[1].iov_len = header->size_compressed;
657  iov[1].iov_base = compressed;
658 
659  free(buffer);
660 
661  biggest = QB_MAX(header->size_compressed, biggest);
662 
663  } else {
664  ssize_t rc = -EMSGSIZE;
665 
666  crm_log_xml_trace(message, "EMSGSIZE");
667  biggest = QB_MAX(header->size_uncompressed, biggest);
668 
669  crm_err
670  ("Could not compress the message (%u bytes) into less than the configured ipc limit (%u bytes). "
671  "Set PCMK_ipc_buffer to a higher value (%u bytes suggested)",
672  header->size_uncompressed, max_send_size, 4 * biggest);
673 
674  free(compressed);
675  free(buffer);
676  free(header);
677  free(iov);
678 
679  return rc;
680  }
681  }
682 
683  header->qb.size = iov[0].iov_len + iov[1].iov_len;
684  header->qb.id = (int32_t)request; /* Replying to a specific request */
685 
686  *result = iov;
687  CRM_ASSERT(header->qb.size > 0);
688  return header->qb.size;
689 }
690 
691 ssize_t
692 crm_ipcs_sendv(crm_client_t * c, struct iovec * iov, enum crm_ipc_flags flags)
693 {
694  ssize_t rc;
695  static uint32_t id = 1;
696  struct crm_ipc_response_header *header = iov[0].iov_base;
697 
699  /* _ALL_ replies to proxied connections need to be sent as events */
700  if (is_not_set(flags, crm_ipc_server_event)) {
701  flags |= crm_ipc_server_event;
702  /* this flag lets us know this was originally meant to be a response.
703  * even though we're sending it over the event channel. */
705  }
706  }
707 
708  header->flags |= flags;
709  if (flags & crm_ipc_server_event) {
710  header->qb.id = id++; /* We don't really use it, but doesn't hurt to set one */
711 
712  if (flags & crm_ipc_server_free) {
713  crm_trace("Sending the original to %p[%d]", c->ipcs, c->pid);
714  c->event_queue = g_list_append(c->event_queue, iov);
715 
716  } else {
717  struct iovec *iov_copy = calloc(2, sizeof(struct iovec));
718 
719  crm_trace("Sending a copy to %p[%d]", c->ipcs, c->pid);
720  iov_copy[0].iov_len = iov[0].iov_len;
721  iov_copy[0].iov_base = malloc(iov[0].iov_len);
722  memcpy(iov_copy[0].iov_base, iov[0].iov_base, iov[0].iov_len);
723 
724  iov_copy[1].iov_len = iov[1].iov_len;
725  iov_copy[1].iov_base = malloc(iov[1].iov_len);
726  memcpy(iov_copy[1].iov_base, iov[1].iov_base, iov[1].iov_len);
727 
728  c->event_queue = g_list_append(c->event_queue, iov_copy);
729  }
730 
731  } else {
732  CRM_LOG_ASSERT(header->qb.id != 0); /* Replying to a specific request */
733 
734  rc = qb_ipcs_response_sendv(c->ipcs, iov, 2);
735  if (rc < header->qb.size) {
736  crm_notice("Response %d to %p[%d] (%u bytes) failed: %s (%d)",
737  header->qb.id, c->ipcs, c->pid, header->qb.size, pcmk_strerror(rc), rc);
738 
739  } else {
740  crm_trace("Response %d sent, %lld bytes to %p[%d]",
741  header->qb.id, (long long) rc, c->ipcs, c->pid);
742  }
743 
744  if (flags & crm_ipc_server_free) {
745  free(iov[0].iov_base);
746  free(iov[1].iov_base);
747  free(iov);
748  }
749  }
750 
751  if (flags & crm_ipc_server_event) {
752  rc = crm_ipcs_flush_events(c);
753  } else {
755  }
756 
757  if (rc == -EPIPE || rc == -ENOTCONN) {
758  crm_trace("Client %p disconnected", c->ipcs);
759  }
760 
761  return rc;
762 }
763 
764 ssize_t
765 crm_ipcs_send(crm_client_t * c, uint32_t request, xmlNode * message,
766  enum crm_ipc_flags flags)
767 {
768  struct iovec *iov = NULL;
769  ssize_t rc = 0;
770 
771  if(c == NULL) {
772  return -EDESTADDRREQ;
773  }
774  crm_ipc_init();
775 
776  rc = crm_ipc_prepare(request, message, &iov, ipc_buffer_max);
777  if (rc > 0) {
778  rc = crm_ipcs_sendv(c, iov, flags | crm_ipc_server_free);
779 
780  } else {
781  free(iov);
782  crm_notice("Message to %p[%d] failed: %s (%d)",
783  c->ipcs, c->pid, pcmk_strerror(rc), rc);
784  }
785 
786  return rc;
787 }
788 
789 void
790 crm_ipcs_send_ack(crm_client_t * c, uint32_t request, uint32_t flags, const char *tag, const char *function,
791  int line)
792 {
793  if (flags & crm_ipc_client_response) {
794  xmlNode *ack = create_xml_node(NULL, tag);
795 
796  crm_trace("Ack'ing msg from %s (%p)", crm_client_name(c), c);
797  c->request_id = 0;
798  crm_xml_add(ack, "function", function);
799  crm_xml_add_int(ack, "line", line);
800  crm_ipcs_send(c, request, ack, flags);
801  free_xml(ack);
802  }
803 }
804 
805 /* Client... */
806 
807 #define MIN_MSG_SIZE 12336 /* sizeof(struct qb_ipc_connection_response) */
808 #define MAX_MSG_SIZE 128*1024 /* 128k default */
809 
810 struct crm_ipc_s {
811  struct pollfd pfd;
812 
813  /* the max size we can send/receive over ipc */
814  unsigned int max_buf_size;
815  /* Size of the allocated 'buffer' */
816  unsigned int buf_size;
817  int msg_size;
818  int need_reply;
819  char *buffer;
820  char *name;
821  uint32_t buffer_flags;
822 
823  qb_ipcc_connection_t *ipc;
824 
825 };
826 
827 static unsigned int
828 pick_ipc_buffer(unsigned int max)
829 {
830  static unsigned int global_max = 0;
831 
832  if (global_max == 0) {
833  const char *env = getenv("PCMK_ipc_buffer");
834 
835  if (env) {
836  int env_max = crm_parse_int(env, "0");
837 
838  global_max = (env_max > 0)? QB_MAX(MIN_MSG_SIZE, env_max) : MAX_MSG_SIZE;
839 
840  } else {
841  global_max = MAX_MSG_SIZE;
842  }
843  }
844 
845  return QB_MAX(max, global_max);
846 }
847 
848 crm_ipc_t *
849 crm_ipc_new(const char *name, size_t max_size)
850 {
851  crm_ipc_t *client = NULL;
852 
853  client = calloc(1, sizeof(crm_ipc_t));
854 
855  client->name = strdup(name);
856  client->buf_size = pick_ipc_buffer(max_size);
857  client->buffer = malloc(client->buf_size);
858 
859  /* Clients initiating connection pick the max buf size */
860  client->max_buf_size = client->buf_size;
861 
862  client->pfd.fd = -1;
863  client->pfd.events = POLLIN;
864  client->pfd.revents = 0;
865 
866  return client;
867 }
868 
878 bool
880 {
881  static uid_t cl_uid = 0;
882  static gid_t cl_gid = 0;
883  pid_t found_pid = 0; uid_t found_uid = 0; gid_t found_gid = 0;
884  int rv;
885 
886  client->need_reply = FALSE;
887  client->ipc = qb_ipcc_connect(client->name, client->buf_size);
888 
889  if (client->ipc == NULL) {
890  crm_debug("Could not establish %s connection: %s (%d)", client->name, pcmk_strerror(errno), errno);
891  return FALSE;
892  }
893 
894  client->pfd.fd = crm_ipc_get_fd(client);
895  if (client->pfd.fd < 0) {
896  rv = errno;
897  /* message already omitted */
898  crm_ipc_close(client);
899  errno = rv;
900  return FALSE;
901  }
902 
903  if (!cl_uid && !cl_gid
904  && (rv = crm_user_lookup(CRM_DAEMON_USER, &cl_uid, &cl_gid)) < 0) {
905  errno = -rv;
906  /* message already omitted */
907  crm_ipc_close(client);
908  errno = -rv;
909  return FALSE;
910  }
911 
912  if (!(rv = crm_ipc_is_authentic_process(client->pfd.fd, cl_uid, cl_gid,
913  &found_pid, &found_uid,
914  &found_gid))) {
915  crm_err("Daemon (IPC %s) is not authentic:"
916  " process %lld (uid: %lld, gid: %lld)",
917  client->name, (long long) PCMK__SPECIAL_PID_AS_0(found_pid),
918  (long long) found_uid, (long long) found_gid);
919  crm_ipc_close(client);
920  errno = ECONNABORTED;
921  return FALSE;
922 
923  } else if (rv < 0) {
924  errno = -rv;
925  crm_perror(LOG_ERR, "Could not verify authenticity of daemon (IPC %s)",
926  client->name);
927  crm_ipc_close(client);
928  errno = -rv;
929  return FALSE;
930  }
931 
932  qb_ipcc_context_set(client->ipc, client);
933 
934 #ifdef HAVE_IPCS_GET_BUFFER_SIZE
935  client->max_buf_size = qb_ipcc_get_buffer_size(client->ipc);
936  if (client->max_buf_size > client->buf_size) {
937  free(client->buffer);
938  client->buffer = calloc(1, client->max_buf_size);
939  client->buf_size = client->max_buf_size;
940  }
941 #endif
942 
943  return TRUE;
944 }
945 
946 void
948 {
949  if (client) {
950  crm_trace("Disconnecting %s IPC connection %p (%p)", client->name, client, client->ipc);
951 
952  if (client->ipc) {
953  qb_ipcc_connection_t *ipc = client->ipc;
954 
955  client->ipc = NULL;
956  qb_ipcc_disconnect(ipc);
957  }
958  }
959 }
960 
961 void
963 {
964  if (client) {
965  if (client->ipc && qb_ipcc_is_connected(client->ipc)) {
966  crm_notice("Destroying an active IPC connection to %s", client->name);
967  /* The next line is basically unsafe
968  *
969  * If this connection was attached to mainloop and mainloop is active,
970  * the 'disconnected' callback will end up back here and we'll end
971  * up free'ing the memory twice - something that can still happen
972  * even without this if we destroy a connection and it closes before
973  * we call exit
974  */
975  /* crm_ipc_close(client); */
976  }
977  crm_trace("Destroying IPC connection to %s: %p", client->name, client);
978  free(client->buffer);
979  free(client->name);
980  free(client);
981  }
982 }
983 
984 int
986 {
987  int fd = 0;
988 
989  if (client && client->ipc && (qb_ipcc_fd_get(client->ipc, &fd) == 0)) {
990  return fd;
991  }
992  errno = EINVAL;
993  crm_perror(LOG_ERR, "Could not obtain file IPC descriptor for %s",
994  (client? client->name : "unspecified client"));
995  return -errno;
996 }
997 
998 bool
1000 {
1001  bool rc = FALSE;
1002 
1003  if (client == NULL) {
1004  crm_trace("No client");
1005  return FALSE;
1006 
1007  } else if (client->ipc == NULL) {
1008  crm_trace("No connection");
1009  return FALSE;
1010 
1011  } else if (client->pfd.fd < 0) {
1012  crm_trace("Bad descriptor");
1013  return FALSE;
1014  }
1015 
1016  rc = qb_ipcc_is_connected(client->ipc);
1017  if (rc == FALSE) {
1018  client->pfd.fd = -EINVAL;
1019  }
1020  return rc;
1021 }
1022 
1030 int
1032 {
1033  int rc;
1034 
1035  CRM_ASSERT(client != NULL);
1036 
1037  if (crm_ipc_connected(client) == FALSE) {
1038  return -ENOTCONN;
1039  }
1040 
1041  client->pfd.revents = 0;
1042  rc = poll(&(client->pfd), 1, 0);
1043  return (rc < 0)? -errno : rc;
1044 }
1045 
1046 static int
1047 crm_ipc_decompress(crm_ipc_t * client)
1048 {
1049  struct crm_ipc_response_header *header = (struct crm_ipc_response_header *)(void*)client->buffer;
1050 
1051  if (header->size_compressed) {
1052  int rc = 0;
1053  unsigned int size_u = 1 + header->size_uncompressed;
1054  /* never let buf size fall below our max size required for ipc reads. */
1055  unsigned int new_buf_size = QB_MAX((hdr_offset + size_u), client->max_buf_size);
1056  char *uncompressed = calloc(1, new_buf_size);
1057 
1058  crm_trace("Decompressing message data %u bytes into %u bytes",
1059  header->size_compressed, size_u);
1060 
1061  rc = BZ2_bzBuffToBuffDecompress(uncompressed + hdr_offset, &size_u,
1062  client->buffer + hdr_offset, header->size_compressed, 1, 0);
1063 
1064  if (rc != BZ_OK) {
1065  crm_err("Decompression failed: %s (%d)", bz2_strerror(rc), rc);
1066  free(uncompressed);
1067  return -EILSEQ;
1068  }
1069 
1070  /*
1071  * This assert no longer holds true. For an identical msg, some clients may
1072  * require compression, and others may not. If that same msg (event) is sent
1073  * to multiple clients, it could result in some clients receiving a compressed
1074  * msg even though compression was not explicitly required for them.
1075  *
1076  * CRM_ASSERT((header->size_uncompressed + hdr_offset) >= ipc_buffer_max);
1077  */
1078  CRM_ASSERT(size_u == header->size_uncompressed);
1079 
1080  memcpy(uncompressed, client->buffer, hdr_offset); /* Preserve the header */
1081  header = (struct crm_ipc_response_header *)(void*)uncompressed;
1082 
1083  free(client->buffer);
1084  client->buf_size = new_buf_size;
1085  client->buffer = uncompressed;
1086  }
1087 
1088  CRM_ASSERT(client->buffer[hdr_offset + header->size_uncompressed - 1] == 0);
1089  return pcmk_ok;
1090 }
1091 
1092 long
1094 {
1095  struct crm_ipc_response_header *header = NULL;
1096 
1097  CRM_ASSERT(client != NULL);
1098  CRM_ASSERT(client->ipc != NULL);
1099  CRM_ASSERT(client->buffer != NULL);
1100 
1101  crm_ipc_init();
1102 
1103  client->buffer[0] = 0;
1104  client->msg_size = qb_ipcc_event_recv(client->ipc, client->buffer, client->buf_size - 1, 0);
1105  if (client->msg_size >= 0) {
1106  int rc = crm_ipc_decompress(client);
1107 
1108  if (rc != pcmk_ok) {
1109  return rc;
1110  }
1111 
1112  header = (struct crm_ipc_response_header *)(void*)client->buffer;
1113  if(header->version > PCMK_IPC_VERSION) {
1114  crm_err("Filtering incompatible v%d IPC message, we only support versions <= %d",
1115  header->version, PCMK_IPC_VERSION);
1116  return -EBADMSG;
1117  }
1118 
1119  crm_trace("Received %s event %d, size=%u, rc=%d, text: %.100s",
1120  client->name, header->qb.id, header->qb.size, client->msg_size,
1121  client->buffer + hdr_offset);
1122 
1123  } else {
1124  crm_trace("No message from %s received: %s", client->name, pcmk_strerror(client->msg_size));
1125  }
1126 
1127  if (crm_ipc_connected(client) == FALSE || client->msg_size == -ENOTCONN) {
1128  crm_err("Connection to %s failed", client->name);
1129  }
1130 
1131  if (header) {
1132  /* Data excluding the header */
1133  return header->size_uncompressed;
1134  }
1135  return -ENOMSG;
1136 }
1137 
1138 const char *
1140 {
1141  CRM_ASSERT(client != NULL);
1142  return client->buffer + sizeof(struct crm_ipc_response_header);
1143 }
1144 
1145 uint32_t
1147 {
1148  struct crm_ipc_response_header *header = NULL;
1149 
1150  CRM_ASSERT(client != NULL);
1151  if (client->buffer == NULL) {
1152  return 0;
1153  }
1154 
1155  header = (struct crm_ipc_response_header *)(void*)client->buffer;
1156  return header->flags;
1157 }
1158 
1159 const char *
1161 {
1162  CRM_ASSERT(client != NULL);
1163  return client->name;
1164 }
1165 
1166 static int
1167 internal_ipc_send_recv(crm_ipc_t * client, const void *iov)
1168 {
1169  int rc = 0;
1170 
1171  do {
1172  rc = qb_ipcc_sendv_recv(client->ipc, iov, 2, client->buffer, client->buf_size, -1);
1173  } while (rc == -EAGAIN && crm_ipc_connected(client));
1174 
1175  return rc;
1176 }
1177 
1178 static int
1179 internal_ipc_send_request(crm_ipc_t * client, const void *iov, int ms_timeout)
1180 {
1181  int rc = 0;
1182  time_t timeout = time(NULL) + 1 + (ms_timeout / 1000);
1183 
1184  do {
1185  rc = qb_ipcc_sendv(client->ipc, iov, 2);
1186  } while (rc == -EAGAIN && time(NULL) < timeout && crm_ipc_connected(client));
1187 
1188  return rc;
1189 }
1190 
1191 static int
1192 internal_ipc_get_reply(crm_ipc_t * client, int request_id, int ms_timeout)
1193 {
1194  time_t timeout = time(NULL) + 1 + (ms_timeout / 1000);
1195  int rc = 0;
1196 
1197  crm_ipc_init();
1198 
1199  /* get the reply */
1200  crm_trace("client %s waiting on reply to msg id %d", client->name, request_id);
1201  do {
1202 
1203  rc = qb_ipcc_recv(client->ipc, client->buffer, client->buf_size, 1000);
1204  if (rc > 0) {
1205  struct crm_ipc_response_header *hdr = NULL;
1206 
1207  int rc = crm_ipc_decompress(client);
1208 
1209  if (rc != pcmk_ok) {
1210  return rc;
1211  }
1212 
1213  hdr = (struct crm_ipc_response_header *)(void*)client->buffer;
1214  if (hdr->qb.id == request_id) {
1215  /* Got it */
1216  break;
1217  } else if (hdr->qb.id < request_id) {
1218  xmlNode *bad = string2xml(crm_ipc_buffer(client));
1219 
1220  crm_err("Discarding old reply %d (need %d)", hdr->qb.id, request_id);
1221  crm_log_xml_notice(bad, "OldIpcReply");
1222 
1223  } else {
1224  xmlNode *bad = string2xml(crm_ipc_buffer(client));
1225 
1226  crm_err("Discarding newer reply %d (need %d)", hdr->qb.id, request_id);
1227  crm_log_xml_notice(bad, "ImpossibleReply");
1228  CRM_ASSERT(hdr->qb.id <= request_id);
1229  }
1230  } else if (crm_ipc_connected(client) == FALSE) {
1231  crm_err("Server disconnected client %s while waiting for msg id %d", client->name,
1232  request_id);
1233  break;
1234  }
1235 
1236  } while (time(NULL) < timeout);
1237 
1238  return rc;
1239 }
1240 
1241 int
1242 crm_ipc_send(crm_ipc_t * client, xmlNode * message, enum crm_ipc_flags flags, int32_t ms_timeout,
1243  xmlNode ** reply)
1244 {
1245  long rc = 0;
1246  struct iovec *iov;
1247  static uint32_t id = 0;
1248  static int factor = 8;
1249  struct crm_ipc_response_header *header;
1250 
1251  crm_ipc_init();
1252 
1253  if (client == NULL) {
1254  crm_notice("Invalid connection");
1255  return -ENOTCONN;
1256 
1257  } else if (crm_ipc_connected(client) == FALSE) {
1258  /* Don't even bother */
1259  crm_notice("Connection to %s closed", client->name);
1260  return -ENOTCONN;
1261  }
1262 
1263  if (ms_timeout == 0) {
1264  ms_timeout = 5000;
1265  }
1266 
1267  if (client->need_reply) {
1268  crm_trace("Trying again to obtain pending reply from %s", client->name);
1269  rc = qb_ipcc_recv(client->ipc, client->buffer, client->buf_size, ms_timeout);
1270  if (rc < 0) {
1271  crm_warn("Sending to %s (%p) is disabled until pending reply is received", client->name,
1272  client->ipc);
1273  return -EALREADY;
1274 
1275  } else {
1276  crm_notice("Lost reply from %s (%p) finally arrived, sending re-enabled", client->name,
1277  client->ipc);
1278  client->need_reply = FALSE;
1279  }
1280  }
1281 
1282  id++;
1283  CRM_LOG_ASSERT(id != 0); /* Crude wrap-around detection */
1284  rc = crm_ipc_prepare(id, message, &iov, client->max_buf_size);
1285  if(rc < 0) {
1286  return rc;
1287  }
1288 
1289  header = iov[0].iov_base;
1290  header->flags |= flags;
1291 
1292  if(is_set(flags, crm_ipc_proxied)) {
1293  /* Don't look for a synchronous response */
1295  }
1296 
1297  if(header->size_compressed) {
1298  if(factor < 10 && (client->max_buf_size / 10) < (rc / factor)) {
1299  crm_notice("Compressed message exceeds %d0%% of the configured ipc limit (%u bytes), "
1300  "consider setting PCMK_ipc_buffer to %u or higher",
1301  factor, client->max_buf_size, 2 * client->max_buf_size);
1302  factor++;
1303  }
1304  }
1305 
1306  crm_trace("Sending from client: %s request id: %d bytes: %u timeout:%d msg...",
1307  client->name, header->qb.id, header->qb.size, ms_timeout);
1308 
1309  if (ms_timeout > 0 || is_not_set(flags, crm_ipc_client_response)) {
1310 
1311  rc = internal_ipc_send_request(client, iov, ms_timeout);
1312 
1313  if (rc <= 0) {
1314  crm_trace("Failed to send from client %s request %d with %u bytes...",
1315  client->name, header->qb.id, header->qb.size);
1316  goto send_cleanup;
1317 
1318  } else if (is_not_set(flags, crm_ipc_client_response)) {
1319  crm_trace("Message sent, not waiting for reply to %d from %s to %u bytes...",
1320  header->qb.id, client->name, header->qb.size);
1321 
1322  goto send_cleanup;
1323  }
1324 
1325  rc = internal_ipc_get_reply(client, header->qb.id, ms_timeout);
1326  if (rc < 0) {
1327  /* No reply, for now, disable sending
1328  *
1329  * The alternative is to close the connection since we don't know
1330  * how to detect and discard out-of-sequence replies
1331  *
1332  * TODO - implement the above
1333  */
1334  client->need_reply = TRUE;
1335  }
1336 
1337  } else {
1338  rc = internal_ipc_send_recv(client, iov);
1339  }
1340 
1341  if (rc > 0) {
1342  struct crm_ipc_response_header *hdr = (struct crm_ipc_response_header *)(void*)client->buffer;
1343 
1344  crm_trace("Received response %d, size=%u, rc=%ld, text: %.200s", hdr->qb.id, hdr->qb.size,
1345  rc, crm_ipc_buffer(client));
1346 
1347  if (reply) {
1348  *reply = string2xml(crm_ipc_buffer(client));
1349  }
1350 
1351  } else {
1352  crm_trace("Response not received: rc=%ld, errno=%d", rc, errno);
1353  }
1354 
1355  send_cleanup:
1356  if (crm_ipc_connected(client) == FALSE) {
1357  crm_notice("Connection to %s closed: %s (%ld)", client->name, pcmk_strerror(rc), rc);
1358 
1359  } else if (rc == -ETIMEDOUT) {
1360  crm_warn("Request %d to %s (%p) failed: %s (%ld) after %dms",
1361  header->qb.id, client->name, client->ipc, pcmk_strerror(rc), rc, ms_timeout);
1362  crm_write_blackbox(0, NULL);
1363 
1364  } else if (rc <= 0) {
1365  crm_warn("Request %d to %s (%p) failed: %s (%ld)",
1366  header->qb.id, client->name, client->ipc, pcmk_strerror(rc), rc);
1367  }
1368 
1369  free(header);
1370  free(iov[1].iov_base);
1371  free(iov);
1372  return rc;
1373 }
1374 
1375 int
1376 crm_ipc_is_authentic_process(int sock, uid_t refuid, gid_t refgid,
1377  pid_t *gotpid, uid_t *gotuid, gid_t *gotgid) {
1378  int ret = 0;
1379  pid_t found_pid = 0; uid_t found_uid = 0; gid_t found_gid = 0;
1380 #if defined(US_AUTH_PEERCRED_UCRED)
1381  struct ucred ucred;
1382  socklen_t ucred_len = sizeof(ucred);
1383 
1384  if (!getsockopt(sock, SOL_SOCKET, SO_PEERCRED,
1385  &ucred, &ucred_len)
1386  && ucred_len == sizeof(ucred)) {
1387  found_pid = ucred.pid; found_uid = ucred.uid; found_gid = ucred.gid;
1388 
1389 #elif defined(US_AUTH_PEERCRED_SOCKPEERCRED)
1390  struct sockpeercred sockpeercred;
1391  socklen_t sockpeercred_len = sizeof(sockpeercred);
1392 
1393  if (!getsockopt(sock, SOL_SOCKET, SO_PEERCRED,
1394  &sockpeercred, &sockpeercred_len)
1395  && sockpeercred_len == sizeof(sockpeercred_len)) {
1396  found_pid = sockpeercred.pid;
1397  found_uid = sockpeercred.uid; found_gid = sockpeercred.gid;
1398 
1399 #elif defined(US_AUTH_GETPEEREID)
1400  if (!getpeereid(sock, &found_uid, &found_gid)) {
1401  found_pid = PCMK__SPECIAL_PID; /* cannot obtain PID (FreeBSD) */
1402 
1403 #elif defined(US_AUTH_GETPEERUCRED)
1404  ucred_t *ucred;
1405  if (!getpeerucred(sock, &ucred)) {
1406  errno = 0;
1407  found_pid = ucred_getpid(ucred);
1408  found_uid = ucred_geteuid(ucred); found_gid = ucred_getegid(ucred);
1409  ret = -errno;
1410  ucred_free(ucred);
1411  if (ret) {
1412  return (ret < 0) ? ret : -pcmk_err_generic;
1413  }
1414 
1415 #else
1416 # error "No way to authenticate a Unix socket peer"
1417  errno = 0;
1418  if (0) {
1419 #endif
1420  if (gotpid != NULL) {
1421  *gotpid = found_pid;
1422  }
1423  if (gotuid != NULL) {
1424  *gotuid = found_uid;
1425  }
1426  if (gotgid != NULL) {
1427  *gotgid = found_gid;
1428  }
1429  ret = (found_uid == 0 || found_uid == refuid || found_gid == refgid);
1430  } else {
1431  ret = (errno > 0) ? -errno : -pcmk_err_generic;
1432  }
1433 
1434  return ret;
1435 }
1436 
1437 int
1438 pcmk__ipc_is_authentic_process_active(const char *name, uid_t refuid,
1439  gid_t refgid, pid_t *gotpid) {
1440  static char last_asked_name[PATH_MAX / 2] = ""; /* log spam prevention */
1441  int fd, ret = 0;
1442  pid_t found_pid = 0; uid_t found_uid = 0; gid_t found_gid = 0;
1443  qb_ipcc_connection_t *c;
1444 
1445  if ((c = qb_ipcc_connect(name, 0)) == NULL) {
1446  crm_info("Could not connect to %s IPC: %s", name, strerror(errno));
1447 
1448  } else if ((ret = qb_ipcc_fd_get(c, &fd))) {
1449  crm_err("Could not get fd from %s IPC: %s (%d)", name,
1450  strerror(-ret), -ret);
1451  ret = -1;
1452 
1453  } else if ((ret = crm_ipc_is_authentic_process(fd, refuid, refgid,
1454  &found_pid, &found_uid,
1455  &found_gid)) < 0) {
1456  if (ret == -pcmk_err_generic) {
1457  crm_err("Could not get peer credentials from %s IPC", name);
1458  } else {
1459  crm_err("Could not get peer credentials from %s IPC: %s (%d)",
1460  name, strerror(-ret), -ret);
1461  }
1462  ret = -1;
1463 
1464  } else {
1465  if (gotpid != NULL) {
1466  *gotpid = found_pid;
1467  }
1468 
1469  if (!ret) {
1470  crm_err("Daemon (IPC %s) effectively blocked with unauthorized"
1471  " process %lld (uid: %lld, gid: %lld)",
1472  name, (long long) PCMK__SPECIAL_PID_AS_0(found_pid),
1473  (long long) found_uid, (long long) found_gid);
1474  ret = -2;
1475  } else if ((found_uid != refuid || found_gid != refgid)
1476  && strncmp(last_asked_name, name, sizeof(last_asked_name))) {
1477  if (!found_uid && refuid) {
1478  crm_warn("Daemon (IPC %s) runs as root, whereas the expected"
1479  " credentials are %lld:%lld, hazard of violating"
1480  " the least privilege principle",
1481  name, (long long) refuid, (long long) refgid);
1482  } else {
1483  crm_notice("Daemon (IPC %s) runs as %lld:%lld, whereas the"
1484  " expected credentials are %lld:%lld, which may"
1485  " mean a different set of privileges than expected",
1486  name, (long long) found_uid, (long long) found_gid,
1487  (long long) refuid, (long long) refgid);
1488  }
1489  memccpy(last_asked_name, name, '\0', sizeof(last_asked_name));
1490  }
1491  }
1492 
1493  if (ret) { /* here, !ret only when we could not initially connect */
1494  qb_ipcc_disconnect(c);
1495  }
1496 
1497  return ret;
1498 }
1499 
1500 
1501 /* Utils */
1502 
1503 xmlNode *
1504 create_hello_message(const char *uuid,
1505  const char *client_name, const char *major_version, const char *minor_version)
1506 {
1507  xmlNode *hello_node = NULL;
1508  xmlNode *hello = NULL;
1509 
1510  if (uuid == NULL || strlen(uuid) == 0
1511  || client_name == NULL || strlen(client_name) == 0
1512  || major_version == NULL || strlen(major_version) == 0
1513  || minor_version == NULL || strlen(minor_version) == 0) {
1514  crm_err("Missing fields, Hello message will not be valid.");
1515  return NULL;
1516  }
1517 
1518  hello_node = create_xml_node(NULL, XML_TAG_OPTIONS);
1519  crm_xml_add(hello_node, "major_version", major_version);
1520  crm_xml_add(hello_node, "minor_version", minor_version);
1521  crm_xml_add(hello_node, "client_name", client_name);
1522  crm_xml_add(hello_node, "client_uuid", uuid);
1523 
1524  crm_trace("creating hello message");
1525  hello = create_request(CRM_OP_HELLO, hello_node, NULL, NULL, client_name, uuid);
1526  free_xml(hello_node);
1527 
1528  return hello;
1529 }
#define F_CRM_TASK
Definition: msg_xml.h:56
const char * crm_ipc_buffer(crm_ipc_t *client)
Definition: ipc.c:1139
#define F_CRM_REFERENCE
Definition: msg_xml.h:62
void crm_write_blackbox(int nsig, struct qb_log_callsite *callsite)
Definition: logging.c:426
A dumping ground.
#define F_TYPE
Definition: msg_xml.h:34
void crm_ipc_close(crm_ipc_t *client)
Definition: ipc.c:947
#define crm_notice(fmt, args...)
Definition: logging.h:250
char * crm_generate_uuid(void)
Definition: utils.c:1398
uint32_t crm_ipc_buffer_flags(crm_ipc_t *client)
Definition: ipc.c:1146
int crm_ipc_send(crm_ipc_t *client, xmlNode *message, enum crm_ipc_flags flags, int32_t ms_timeout, xmlNode **reply)
Definition: ipc.c:1242
#define F_CRM_HOST_TO
Definition: msg_xml.h:57
#define XML_TAG_OPTIONS
Definition: msg_xml.h:122
const char * pcmk_strerror(int rc)
Definition: logging.c:1135
crm_client_t * crm_client_get(qb_ipcs_connection_t *c)
Definition: ipc.c:211
uint32_t flags
Definition: ipcs.h:83
qb_ipcs_connection_t * ipcs
Definition: ipcs.h:94
#define F_CRM_MSG_TYPE
Definition: msg_xml.h:58
uint32_t size
Definition: internal.h:52
int request_id
Definition: ipcs.h:82
#define CRM_FEATURE_SET
Definition: crm.h:36
#define F_CRM_HOST_FROM
Definition: msg_xml.h:61
#define pcmk_ok
Definition: error.h:42
#define T_CRM
Definition: msg_xml.h:46
long long crm_int_helper(const char *text, char **end_text)
Definition: strings.c:80
bool crm_set_client_queue_max(crm_client_t *client, const char *qmax)
Raise IPC eviction threshold for a client, if allowed.
Definition: ipc.c:425
crm_client_t * crm_client_get_by_id(const char *id)
Definition: ipc.c:222
xmlNode * create_reply_adv(xmlNode *original_request, xmlNode *xml_response_data, const char *origin)
Definition: ipc.c:154
char * buffer
Definition: ipcs.h:44
unsigned int queue_max
Definition: ipcs.h:99
#define PCMK__SPECIAL_PID_AS_0(p)
Definition: ipc_internal.h:34
bool crm_ipc_connect(crm_ipc_t *client)
Establish an IPC connection to a Pacemaker component.
Definition: ipc.c:879
int crm_parse_int(const char *text, const char *default_text)
Definition: strings.c:125
#define PCMK_IPC_VERSION
Definition: ipc.c:42
#define PCMK__SPECIAL_PID
Definition: ipc_internal.h:25
struct crm_remote_s * remote
Definition: ipcs.h:96
int crm_user_lookup(const char *name, uid_t *uid, gid_t *gid)
Definition: utils.c:424
#define CRM_LOG_ASSERT(expr)
Definition: logging.h:150
void crm_client_destroy(crm_client_t *c)
Definition: ipc.c:370
int crm_ipc_get_fd(crm_ipc_t *client)
Definition: ipc.c:985
void crm_client_init(void)
Definition: ipc.c:256
int crm_ipc_is_authentic_process(int sock, uid_t refuid, gid_t refgid, pid_t *gotpid, uid_t *gotuid, gid_t *gotgid)
Check the authenticity of the IPC socket peer process.
Definition: ipc.c:1376
#define clear_bit(word, bit)
Definition: crm_internal.h:212
void crm_client_disconnect_all(qb_ipcs_service_t *service)
Definition: ipc.c:278
char * strerror(int errnum)
ssize_t crm_ipc_prepare(uint32_t request, xmlNode *message, struct iovec **result, uint32_t max_send_size)
Definition: ipc.c:613
char * user
Definition: ipcs.h:76
ssize_t crm_ipcs_flush_events(crm_client_t *c)
Definition: ipc.c:532
xmlNode * string2xml(const char *input)
Definition: xml.c:2750
char version[256]
Definition: plugin.c:84
#define MAX_MSG_SIZE
Definition: ipc.c:808
#define XML_ATTR_REQUEST
Definition: msg_xml.h:127
ssize_t crm_ipcs_sendv(crm_client_t *c, struct iovec *iov, enum crm_ipc_flags flags)
Definition: ipc.c:692
#define crm_warn(fmt, args...)
Definition: logging.h:249
crm_client_t * crm_client_new(qb_ipcs_connection_t *c, uid_t uid_client, gid_t gid_client)
Definition: ipc.c:319
#define set_bit(word, bit)
Definition: crm_internal.h:211
#define crm_debug(fmt, args...)
Definition: logging.h:253
#define F_CRM_SYS_TO
Definition: msg_xml.h:59
struct crm_ipc_s crm_ipc_t
Definition: ipc.h:52
const char * crm_ipc_name(crm_ipc_t *client)
Definition: ipc.c:1160
GList * event_queue
Definition: ipcs.h:87
GHashTable * client_connections
Definition: ipc.c:208
unsigned int crm_ipc_default_buffer_size(void)
Definition: ipc.c:71
#define crm_trace(fmt, args...)
Definition: logging.h:254
void crm_ipc_destroy(crm_ipc_t *client)
Definition: ipc.c:962
xmlNode * create_xml_node(xmlNode *parent, const char *name)
Definition: xml.c:2588
const char * crm_element_value(xmlNode *data, const char *name)
Definition: xml.c:5165
#define pcmk_err_generic
Definition: error.h:45
#define CRM_DAEMON_USER
Definition: config.h:47
gboolean add_message_xml(xmlNode *msg, const char *field, xmlNode *xml)
Definition: xml.c:3171
void free_xml(xmlNode *child)
Definition: xml.c:2706
void crm_ipcs_send_ack(crm_client_t *c, uint32_t request, uint32_t flags, const char *tag, const char *function, int line)
Definition: ipc.c:790
unsigned int queue_backlog
Definition: ipcs.h:98
xmlNode * crm_ipcs_recv(crm_client_t *c, void *data, size_t size, uint32_t *id, uint32_t *flags)
Definition: ipc.c:449
int auth_timeout
Definition: ipcs.h:47
#define F_CRM_DATA
Definition: msg_xml.h:55
const char * crm_xml_add(xmlNode *node, const char *name, const char *value)
Definition: xml.c:2490
#define CRM_XS
Definition: logging.h:42
const char * crm_xml_add_int(xmlNode *node, const char *name, int value)
Definition: xml.c:2578
crm_client_t * crm_client_alloc(void *key)
Allocate a new crm_client_t object and generate its ID.
Definition: ipc.c:308
bool crm_compress_string(const char *data, int length, int max, char **result, unsigned int *result_len)
Definition: strings.c:413
ssize_t crm_ipcs_send(crm_client_t *c, uint32_t request, xmlNode *message, enum crm_ipc_flags flags)
Definition: ipc.c:765
#define PCMK_IPC_DEFAULT_QUEUE_MAX
Definition: ipc.c:45
int pcmk__ipc_is_authentic_process_active(const char *name, uid_t refuid, gid_t refgid, pid_t *gotpid)
Definition: ipc.c:1438
uint pid
Definition: ipcs.h:69
int event_timer
Definition: ipcs.h:86
#define crm_perror(level, fmt, args...)
Log a system error message.
Definition: logging.h:226
xmlNode * create_hello_message(const char *uuid, const char *client_name, const char *major_version, const char *minor_version)
Definition: ipc.c:1504
#define CRM_OP_HELLO
Definition: crm.h:122
#define crm_err(fmt, args...)
Definition: logging.h:248
const char * bz2_strerror(int rc)
Definition: logging.c:1198
#define F_CRM_SYS_FROM
Definition: msg_xml.h:60
#define crm_log_xml_notice(xml, text)
Definition: logging.h:259
char * dump_xml_unformatted(xmlNode *msg)
Definition: xml.c:3825
int crm_ipc_ready(crm_ipc_t *client)
Check whether an IPC connection is ready to be read.
Definition: ipc.c:1031
#define uint32_t
Definition: stdint.in.h:158
#define XML_ATTR_RESPONSE
Definition: msg_xml.h:128
#define CRM_ASSERT(expr)
Definition: error.h:35
char data[0]
Definition: internal.h:58
long crm_ipc_read(crm_ipc_t *client)
Definition: ipc.c:1093
char * id
Definition: ipcs.h:74
#define uint8_t
Definition: stdint.in.h:144
Wrappers for and extensions to libqb IPC.
char * generate_hash_key(const char *crm_msg_reference, const char *sys)
Definition: utils.c:414
#define F_CRM_ORIGIN
Definition: msg_xml.h:64
int crm_ipcs_client_pid(qb_ipcs_connection_t *c)
Definition: ipc.c:439
#define crm_log_xml_trace(xml, text)
Definition: logging.h:262
void crm_client_cleanup(void)
Definition: ipc.c:265
enum client_type kind
Definition: ipcs.h:92
const char * crm_client_name(crm_client_t *c)
Definition: ipc.c:242
bool crm_ipc_connected(crm_ipc_t *client)
Definition: ipc.c:999
crm_ipc_t * crm_ipc_new(const char *name, size_t max_size)
Definition: ipc.c:849
char * name
Definition: ipcs.h:75
crm_ipc_flags
Definition: ipc.h:32
xmlNode * create_request_adv(const char *task, xmlNode *msg_data, const char *host_to, const char *sys_to, const char *sys_from, const char *uuid_from, const char *origin)
Definition: ipc.c:109
#define create_request(task, xml_data, host_to, sys_to, sys_from, uuid_from)
Definition: ipc.h:25
#define crm_info(fmt, args...)
Definition: logging.h:251
char * uid2username(uid_t uid)
#define F_CRM_VERSION
Definition: msg_xml.h:63
uint64_t flags
Definition: remote.c:156
#define MIN_MSG_SIZE
Definition: ipc.c:807
enum crm_ais_msg_types type
Definition: internal.h:51
#define int32_t
Definition: stdint.in.h:157