data/src/server.c

   1 /*
   2   Copyright (C) 2010 Proxmox Server Solutions GmbH
   3
   4   This program is free software: you can redistribute it and/or modify
   5   it under the terms of the GNU Affero General Public License as published by
   6   the Free Software Foundation, either version 3 of the License, or
   7   (at your option) any later version.
   8
   9   This program is distributed in the hope that it will be useful,
  10   but WITHOUT ANY WARRANTY; without even the implied warranty of
  11   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  12   GNU Affero General Public License for more details.
  13
  14   You should have received a copy of the GNU Affero General Public License
  15   along with this program.  If not, see <http://www.gnu.org/licenses/>.
  16
  17   Author: Dietmar Maurer <dietmar@proxmox.com>
  18
  19 */
  20
  21 #define G_LOG_DOMAIN "ipcs"
  22
  23 #ifdef HAVE_CONFIG_H
  24 #include <config.h>
  25 #endif /* HAVE_CONFIG_H */
  26
  27 #include <stdint.h>
  28 #include <errno.h>
  29 #include <string.h>
  30 #include <sys/syslog.h>
  31 #include <sys/uio.h>
  32
  33 #include <qb/qbdefs.h>
  34 #include <qb/qbutil.h>
  35 #include <qb/qbloop.h>
  36 #include <qb/qbipcs.h>
  37
  38 #include <glib.h>
  39
  40 #include "cfs-utils.h"
  41 #include "cfs-ipc-ops.h"
  42 #include "status.h"
  43 #include "memdb.h"
  44 #include "logger.h"
  45
  46 static GThread *worker;
  47 static qb_loop_t *loop;
  48 static qb_ipcs_service_t* s1;
  49 static GString *outbuf;
  50 static memdb_t *memdb;
  51
  52 static int server_started = 0;   /* protect with server_started_mutex */
  53 static int terminate_server = 0; /* protect with server_started_mutex */
  54 static GCond server_started_cond;
  55 static GCond server_stopped_cond;
  56 static GMutex server_started_mutex;
  57
  58
  59 typedef struct {
  60         struct qb_ipc_request_header req_header;
  61         char name[256];
  62 } cfs_status_update_request_header_t;
  63
  64 typedef struct {
  65         struct qb_ipc_request_header req_header;
  66         char name[256];
  67         char nodename[256];
  68 } cfs_status_get_request_header_t;
  69
  70 typedef struct {
  71         struct qb_ipc_request_header req_header;
  72         uint8_t priority;
  73         uint8_t ident_len;
  74         uint8_t tag_len;
  75         char data[];
  76 } cfs_log_msg_request_header_t;
  77
  78 typedef struct {
  79         struct qb_ipc_request_header req_header;
  80         uint32_t max_entries;
  81         uint32_t res1;
  82         uint32_t res2;
  83         uint32_t res3;
  84 } cfs_log_get_request_header_t;
  85
  86 typedef struct {
  87         struct qb_ipc_request_header req_header;
  88         uint32_t vmid;
  89         char property[];
  90 } cfs_guest_config_propery_get_request_header_t;
  91
  92 typedef struct {
  93         struct qb_ipc_request_header req_header;
  94         char token[];
  95 } cfs_verify_token_request_header_t;
  96
  97 struct s1_context {
  98         int32_t client_pid;
  99         uid_t uid;
 100         gid_t gid;
 101         gboolean read_only;
 102 };
 103
 104 static int32_t s1_connection_accept_fn(
 105         qb_ipcs_connection_t *c,
 106         uid_t uid,
 107         gid_t gid)
 108 {
 109         if ((uid == 0 && gid == 0) || (gid == cfs.gid)) {
 110                 cfs_debug("authenticated connection %d/%d", uid, gid);
 111                 struct s1_context *ctx = g_new0(struct s1_context, 1);
 112                 ctx->uid = uid;
 113                 ctx->gid = gid;
 114                 ctx->read_only = (gid == cfs.gid);
 115
 116                 struct qb_ipcs_connection_stats stats;
 117                 qb_ipcs_connection_stats_get(c, &stats, QB_FALSE);
 118                 ctx->client_pid = stats.client_pid;
 119
 120                 qb_ipcs_context_set(c, ctx);
 121                 return 0;
 122         }
 123         cfs_critical("connection from bad user %d! - rejected", uid);
 124         return 1;
 125 }
 126
 127 static void s1_connection_created_fn(
 128         qb_ipcs_connection_t *c)
 129 {
 130         struct qb_ipcs_stats srv_stats;
 131
 132         qb_ipcs_stats_get(s1, &srv_stats, QB_FALSE);
 133
 134         cfs_debug("Connection created > active:%d > closed:%d",
 135                     srv_stats.active_connections,
 136                     srv_stats.closed_connections);
 137 }
 138
 139 static void s1_connection_destroyed_fn(
 140         qb_ipcs_connection_t *c)
 141 {
 142         cfs_debug("connection about to be freed");
 143
 144         gpointer ctx;
 145         if ((ctx = qb_ipcs_context_get(c)))
 146                 g_free(ctx);
 147
 148 }
 149
 150 static int32_t s1_connection_closed_fn(
 151         qb_ipcs_connection_t *c)
 152 {
 153         struct qb_ipcs_connection_stats stats;
 154
 155         qb_ipcs_connection_stats_get(c, &stats, QB_FALSE);
 156
 157         cfs_debug("Connection to pid:%d destroyed", stats.client_pid);
 158
 159         return 0;
 160 }
 161
 162 static int32_t s1_msg_process_fn(
 163         qb_ipcs_connection_t *c,
 164         void *data,
 165         size_t size)
 166 {
 167         struct qb_ipc_request_header *req_pt =
 168                 (struct qb_ipc_request_header *)data;
 169
 170         struct s1_context *ctx = (struct s1_context *)qb_ipcs_context_get(c);
 171
 172         if (!ctx) {
 173                 cfs_critical("qb_ipcs_context_get failed");
 174                 qb_ipcs_disconnect(c);
 175                 return 0;
 176         }
 177
 178         int32_t request_id __attribute__ ((aligned(8))) = req_pt->id;
 179         int32_t request_size __attribute__ ((aligned(8))) = req_pt->size;
 180         cfs_debug("process msg:%d, size:%d", request_id, request_size);
 181
 182         char *resp = NULL;
 183
 184         g_string_truncate(outbuf, 0);
 185
 186         int32_t result = -ECHRNG;
 187         if (request_id == CFS_IPC_GET_FS_VERSION) {
 188
 189                 if (request_size != sizeof(struct qb_ipc_request_header)) {
 190                         result = -EINVAL;
 191                 } else {
 192                         result = cfs_create_version_msg(outbuf);
 193                 }
 194
 195         } else if (request_id == CFS_IPC_GET_CLUSTER_INFO) {
 196
 197                 if (request_size != sizeof(struct qb_ipc_request_header)) {
 198                         result = -EINVAL;
 199                 } else {
 200                         result = cfs_create_memberlist_msg(outbuf);
 201                 }
 202
 203         } else if (request_id == CFS_IPC_GET_GUEST_LIST) {
 204
 205                 if (request_size != sizeof(struct qb_ipc_request_header)) {
 206                         result = -EINVAL;
 207                 } else {
 208                         result = cfs_create_vmlist_msg(outbuf);
 209                 }
 210         } else if (request_id == CFS_IPC_SET_STATUS) {
 211
 212                 cfs_status_update_request_header_t *rh =
 213                         (cfs_status_update_request_header_t *)data;
 214
 215                 int datasize = request_size - sizeof(cfs_status_update_request_header_t);
 216
 217                 if (ctx->read_only) {
 218                         result = -EPERM;
 219                 } else if (datasize < 0) {
 220                         result = -EINVAL;
 221                 } else {
 222                         /* make sure name is 0 terminated */
 223                         rh->name[sizeof(rh->name) - 1] = 0;
 224
 225                         char *dataptr = (char*) data + sizeof(cfs_status_update_request_header_t);
 226
 227                         result = cfs_status_set(rh->name, dataptr, datasize);
 228                 }
 229         } else if (request_id == CFS_IPC_GET_STATUS) {
 230
 231                 cfs_status_get_request_header_t *rh =
 232                         (cfs_status_get_request_header_t *)data;
 233
 234                 int datasize = request_size - sizeof(cfs_status_get_request_header_t);
 235
 236                 if (datasize < 0) {
 237                         result = -EINVAL;
 238                 } else {
 239                         /* make sure all names are 0 terminated */
 240                         rh->name[sizeof(rh->name) - 1] = 0;
 241                         rh->nodename[sizeof(rh->nodename) - 1] = 0;
 242
 243                         result = cfs_create_status_msg(outbuf, rh->nodename, rh->name);
 244                 }
 245         } else if (request_id == CFS_IPC_GET_CONFIG) {
 246
 247                 int pathlen = request_size - sizeof(struct qb_ipc_request_header);
 248
 249                 if (pathlen <= 0) {
 250                         result = -EINVAL;
 251                 } else {
 252                         /* make sure path is 0 terminated */
 253                         ((char *)data)[request_size - 1] = 0;
 254                         char *path = (char*) data + sizeof(struct qb_ipc_request_header);
 255
 256                         if (ctx->read_only &&  path_is_private(path)) {
 257                                 result = -EPERM;
 258                         } else {
 259                                 gpointer tmp = NULL;
 260                                 result = memdb_read(memdb, path, &tmp);
 261                                 if (result > 0) {
 262                                         g_string_append_len(outbuf, tmp, result);
 263                                         g_free(tmp);
 264                                 }
 265                         }
 266                 }
 267         } else if (request_id == CFS_IPC_LOG_CLUSTER_MSG) {
 268
 269                 cfs_log_msg_request_header_t *rh =
 270                         (cfs_log_msg_request_header_t *)data;
 271
 272                 int datasize = request_size - G_STRUCT_OFFSET(cfs_log_msg_request_header_t, data);
 273                 int msg_len = datasize - rh->ident_len - rh->tag_len;
 274
 275                 if (ctx->read_only) {
 276                         result = -EPERM;
 277                 } else if (msg_len  < 1) {
 278                         result = -EINVAL;
 279                 } else {
 280                         char *msg = rh->data;
 281                         if ((msg[rh->ident_len - 1] == 0) &&
 282                             (msg[rh->ident_len + rh->tag_len - 1] == 0) &&
 283                             (((char *)data)[request_size] == 0)) {
 284
 285                                 char *ident = msg;
 286                                 char *tag = msg + rh->ident_len;
 287                                 msg = msg + rh->ident_len + rh->tag_len;
 288
 289                                 time_t ctime = time(NULL);
 290                                 clog_entry_t *entry = (clog_entry_t *)alloca(CLOG_MAX_ENTRY_SIZE);
 291                                 if (clog_pack(entry, cfs.nodename, ident, tag, ctx->client_pid,
 292                                               ctime, rh->priority, msg)) {
 293                                         cfs_cluster_log(entry);
 294                                 }
 295
 296                                 result = 0;
 297
 298                         } else {
 299                                 result = -EINVAL;
 300                         }
 301                 }
 302         } else if (request_id == CFS_IPC_GET_CLUSTER_LOG) {
 303
 304                 cfs_log_get_request_header_t *rh =
 305                         (cfs_log_get_request_header_t *)data;
 306
 307                 int userlen = request_size - sizeof(cfs_log_get_request_header_t);
 308
 309                 if (userlen <= 0) {
 310                         result = -EINVAL;
 311                 } else {
 312                         /* make sure user string is 0 terminated */
 313                         ((char *)data)[request_size - 1] = 0;
 314                         char *user = (char*) data + sizeof(cfs_log_get_request_header_t);
 315
 316                         uint32_t max = rh->max_entries ?  rh->max_entries : 50;
 317                         cfs_cluster_log_dump(outbuf, user, max);
 318                         result = 0;
 319                 }
 320         } else if (request_id == CFS_IPC_GET_RRD_DUMP) {
 321
 322                 if (request_size != sizeof(struct qb_ipc_request_header)) {
 323                         result = -EINVAL;
 324                 } else {
 325                         cfs_rrd_dump(outbuf);
 326                         result = 0;
 327                 }
 328         } else if (request_id == CFS_IPC_GET_GUEST_CONFIG_PROPERTY) {
 329
 330                 cfs_guest_config_propery_get_request_header_t *rh =
 331                         (cfs_guest_config_propery_get_request_header_t *) data;
 332
 333                 int proplen = request_size - G_STRUCT_OFFSET(cfs_guest_config_propery_get_request_header_t, property);
 334
 335                 result = 0;
 336                 if (rh->vmid < 100 && rh->vmid != 0) {
 337                         cfs_debug("vmid out of range %u", rh->vmid);
 338                         result = -EINVAL;
 339                 } else if (rh->vmid >= 100 && !vmlist_vm_exists(rh->vmid)) {
 340                         result = -ENOENT;
 341                 } else if (proplen <= 0) {
 342                         cfs_debug("proplen <= 0, %d", proplen);
 343                         result = -EINVAL;
 344                 } else {
 345                         ((char *)data)[request_size - 1] = 0; // ensure property is 0 terminated
 346
 347                         cfs_debug("cfs_get_guest_config_property: basic valid checked, do request");
 348
 349                         result = cfs_create_guest_conf_property_msg(outbuf, memdb, rh->property, rh->vmid);
 350                 }
 351         } else if (request_id == CFS_IPC_VERIFY_TOKEN) {
 352
 353                 cfs_verify_token_request_header_t *rh = (cfs_verify_token_request_header_t *) data;
 354                 int tokenlen = request_size - G_STRUCT_OFFSET(cfs_verify_token_request_header_t, token) - 1;
 355
 356                 if (tokenlen <= 0) {
 357                         cfs_debug("tokenlen <= 0, %d", tokenlen);
 358                         result = -EINVAL;
 359                 } else if (memchr(rh->token, '\n', tokenlen) != NULL) {
 360                         cfs_debug("token contains newline");
 361                         result = -EINVAL;
 362                 } else if (rh->token[tokenlen] != '\0') {
 363                         cfs_debug("token not NULL-terminated");
 364                         result = -EINVAL;
 365                 } else if (strnlen(rh->token, tokenlen) != tokenlen) {
 366                         cfs_debug("token contains NULL-byte");
 367                         result = -EINVAL;
 368                 } else {
 369                         cfs_debug("cfs_verify_token: basic validity checked, reading token.cfg");
 370                         gpointer tmp = NULL;
 371                         int bytes_read = memdb_read(memdb, "priv/token.cfg", &tmp);
 372                         size_t remaining = bytes_read > 0 ? bytes_read : 0;
 373                         if (tmp != NULL && remaining >= tokenlen) {
 374                                 char *line = (char *) tmp;
 375                                 char *next_line;
 376                                 const char *const end = line + remaining;
 377                                 size_t linelen;
 378
 379                                 while (line != NULL) {
 380                                         next_line = memchr(line, '\n', remaining);
 381                                         linelen = next_line == NULL ? remaining : next_line - line;
 382                                         if (linelen == tokenlen && strncmp(line, rh->token, linelen) == 0) {
 383                                                 result = 0;
 384                                                 break;
 385                                         }
 386                                         line = next_line;
 387                                         if (line != NULL) {
 388                                                 line += 1;
 389                                                 remaining = end - line;
 390                                         }
 391                                 }
 392                                 if (line == NULL) {
 393                                         result = -ENOENT;
 394                                 }
 395                                 g_free(tmp);
 396                         } else {
 397                                 cfs_debug("token: token.cfg does not exist - ENOENT");
 398                                 result = -ENOENT;
 399                         }
 400                 }
 401         }
 402
 403         cfs_debug("process result %d", result);
 404
 405         if (result >= 0) {
 406                 resp = outbuf->str;
 407                 result = 0;
 408         }
 409
 410         int iov_len = 2;
 411         struct iovec iov[iov_len];
 412         struct qb_ipc_response_header res_header;
 413
 414         int resp_data_len = resp ? outbuf->len : 0;
 415
 416         res_header.id = request_id;
 417         res_header.size = sizeof(res_header) + resp_data_len;
 418         res_header.error = result;
 419
 420         iov[0].iov_base = (char *)&res_header;
 421         iov[0].iov_len = sizeof(res_header);
 422         iov[1].iov_base = resp;
 423         iov[1].iov_len = resp_data_len;
 424
 425         ssize_t res = qb_ipcs_response_sendv(c, iov, iov_len);
 426         if (res < 0) {
 427                 cfs_critical("qb_ipcs_response_send: %s", strerror(errno));
 428                 qb_ipcs_disconnect(c);
 429         }
 430
 431         return 0;
 432 }
 433
 434 static int32_t my_job_add(
 435         enum qb_loop_priority p,
 436         void *data,
 437         qb_loop_job_dispatch_fn fn)
 438 {
 439         return qb_loop_job_add(loop, p, data, fn);
 440 }
 441
 442 static int32_t my_dispatch_add(
 443         enum qb_loop_priority p,
 444         int32_t fd,
 445         int32_t evts,
 446         void *data,
 447         qb_ipcs_dispatch_fn_t fn)
 448 {
 449         return qb_loop_poll_add(loop, p, fd, evts, data, fn);
 450 }
 451
 452 static int32_t my_dispatch_mod(
 453         enum qb_loop_priority p,
 454         int32_t fd,
 455         int32_t evts,
 456         void *data,
 457         qb_ipcs_dispatch_fn_t fn)
 458 {
 459         return qb_loop_poll_mod(loop, p, fd, evts, data, fn);
 460 }
 461
 462 static int32_t my_dispatch_del(
 463         int32_t fd)
 464 {
 465         return qb_loop_poll_del(loop, fd);
 466 }
 467
 468 static struct qb_ipcs_service_handlers service_handlers = {
 469         .connection_accept = s1_connection_accept_fn,
 470         .connection_created = s1_connection_created_fn,
 471         .msg_process = s1_msg_process_fn,
 472         .connection_destroyed = s1_connection_destroyed_fn,
 473         .connection_closed = s1_connection_closed_fn,
 474 };
 475
 476 static struct qb_ipcs_poll_handlers poll_handlers = {
 477         .job_add = my_job_add,
 478         .dispatch_add = my_dispatch_add,
 479         .dispatch_mod = my_dispatch_mod,
 480         .dispatch_del = my_dispatch_del,
 481 };
 482
 483 static void timer_job(void *data)
 484 {
 485         gboolean terminate = FALSE;
 486
 487         g_mutex_lock (&server_started_mutex);
 488
 489         if (terminate_server) {
 490                 cfs_debug ("got terminate request");
 491
 492                 if (loop)
 493                         qb_loop_stop (loop);
 494
 495                 if (s1) {
 496                         qb_ipcs_destroy (s1);
 497                         s1 = 0;
 498                 }
 499                 server_started = 0;
 500
 501                 g_cond_signal (&server_stopped_cond);
 502
 503                 terminate = TRUE;
 504         } else if (!server_started) {
 505                 server_started = 1;
 506                 g_cond_signal (&server_started_cond);
 507         }
 508
 509         g_mutex_unlock (&server_started_mutex);
 510
 511         if (terminate)
 512                 return;
 513
 514         qb_loop_timer_handle th;
 515         qb_loop_timer_add(loop, QB_LOOP_LOW, 1000000000, NULL, timer_job, &th);
 516 }
 517
 518 static gpointer worker_thread(gpointer data)
 519 {
 520         g_return_val_if_fail(loop != NULL, NULL);
 521
 522         cfs_debug("start event loop");
 523
 524         qb_ipcs_run(s1);
 525
 526         qb_loop_timer_handle th;
 527         qb_loop_timer_add(loop, QB_LOOP_LOW, 1000, NULL, timer_job, &th);
 528
 529         qb_loop_run(loop);
 530
 531         cfs_debug("event loop finished - exit worker thread");
 532
 533         return NULL;
 534 }
 535
 536 gboolean server_start(memdb_t *db)
 537 {
 538         g_return_val_if_fail(loop == NULL, FALSE);
 539         g_return_val_if_fail(worker == NULL, FALSE);
 540         g_return_val_if_fail(db != NULL, FALSE);
 541
 542         terminate_server = 0;
 543         server_started = 0;
 544
 545         memdb = db;
 546
 547         outbuf = g_string_sized_new(8192*8);
 548
 549         if (!(loop = qb_loop_create())) {
 550                 cfs_critical("cant create event loop");
 551                 return FALSE;
 552         }
 553
 554         s1 = qb_ipcs_create("pve2", 1, QB_IPC_SHM, &service_handlers);
 555         if (s1 == 0) {
 556                 cfs_critical("qb_ipcs_create failed: %s", strerror(errno));
 557                 return FALSE;
 558         }
 559         qb_ipcs_poll_handlers_set(s1, &poll_handlers);
 560
 561         worker = g_thread_new ("server", worker_thread, NULL);
 562
 563         g_mutex_lock (&server_started_mutex);
 564         while (!server_started)
 565                 g_cond_wait (&server_started_cond, &server_started_mutex);
 566         g_mutex_unlock (&server_started_mutex);
 567
 568         cfs_debug("server started");
 569
 570         return TRUE;
 571 }
 572
 573 void server_stop(void)
 574 {
 575         cfs_debug("server stop");
 576
 577         g_mutex_lock (&server_started_mutex);
 578         terminate_server = 1;
 579         while (server_started)
 580                 g_cond_wait (&server_stopped_cond, &server_started_mutex);
 581         g_mutex_unlock (&server_started_mutex);
 582
 583         if (worker) {
 584                 g_thread_join(worker);
 585                 worker = NULL;
 586         }
 587
 588         cfs_debug("worker thread finished");
 589
 590         if (loop) {
 591                 qb_loop_destroy(loop);
 592
 593                 loop = NULL;
 594         }
 595
 596         if (outbuf) {
 597                 g_string_free(outbuf, TRUE);
 598                 outbuf = NULL;
 599         }
 600 }