[pve-cluster.git] / data / src / server.c

/*
  Copyright (C) 2010 - 2020 Proxmox Server Solutions GmbH

  This program is free software: you can redistribute it and/or modify
  it under the terms of the GNU Affero General Public License as published by
  the Free Software Foundation, either version 3 of the License, or
  (at your option) any later version.

  This program is distributed in the hope that it will be useful,
  but WITHOUT ANY WARRANTY; without even the implied warranty of
  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  GNU Affero General Public License for more details.

  You should have received a copy of the GNU Affero General Public License
  along with this program.  If not, see <http://www.gnu.org/licenses/>.

  Author: Dietmar Maurer <dietmar@proxmox.com>

*/

#define G_LOG_DOMAIN "ipcs"

#ifdef HAVE_CONFIG_H
#include <config.h>
#endif /* HAVE_CONFIG_H */

#include <stdint.h>
#include <errno.h>
#include <string.h>
#include <sys/syslog.h>
#include <sys/uio.h>

#include <qb/qbdefs.h>
#include <qb/qbutil.h>
#include <qb/qbloop.h>
#include <qb/qbipcs.h>

#include <glib.h>

#include "cfs-utils.h"
#include "cfs-ipc-ops.h"
#include "status.h"
#include "memdb.h"
#include "logger.h"

static GThread *worker;
static qb_loop_t *loop;
static qb_ipcs_service_t* s1;
static GString *outbuf;
static memdb_t *memdb;

static int server_started = 0;   /* protect with server_started_mutex */
static int terminate_server = 0; /* protect with server_started_mutex */
static GCond server_started_cond;
static GCond server_stopped_cond;
static GMutex server_started_mutex;


typedef struct {
	struct qb_ipc_request_header req_header;
	char name[256];
} cfs_status_update_request_header_t;

typedef struct {
	struct qb_ipc_request_header req_header;
	char name[256];
	char nodename[256];
} cfs_status_get_request_header_t;

typedef struct {
	struct qb_ipc_request_header req_header;
	uint8_t priority;
	uint8_t ident_len;
	uint8_t tag_len;
	char data[];
} cfs_log_msg_request_header_t;

typedef struct {
	struct qb_ipc_request_header req_header;
	uint32_t max_entries;
	uint32_t res1;
	uint32_t res2;
	uint32_t res3;
} cfs_log_get_request_header_t;

typedef struct {
	struct qb_ipc_request_header req_header;
	uint32_t vmid;
	char property[];
} cfs_guest_config_propery_get_request_header_t;

typedef struct {
	struct qb_ipc_request_header req_header;
	uint32_t vmid;
	uint8_t num_props;
	char props[]; /* list of \0 terminated properties */
} cfs_guest_config_properties_get_request_header_t;

typedef struct {
	struct qb_ipc_request_header req_header;
	char token[];
} cfs_verify_token_request_header_t;

struct s1_context {
	int32_t client_pid;
	uid_t uid;
	gid_t gid;	
	gboolean read_only;
};
 
static int32_t s1_connection_accept_fn(
	qb_ipcs_connection_t *c, 
	uid_t uid, 
	gid_t gid)
{
	if ((uid == 0 && gid == 0) || (gid == cfs.gid)) {
		cfs_debug("authenticated connection %d/%d", uid, gid);
		struct s1_context *ctx = g_new0(struct s1_context, 1);
		ctx->uid = uid;
		ctx->gid = gid;
		ctx->read_only = (gid == cfs.gid);

		struct qb_ipcs_connection_stats stats;
		qb_ipcs_connection_stats_get(c, &stats, QB_FALSE);
		ctx->client_pid = stats.client_pid;

		qb_ipcs_context_set(c, ctx);
		return 0;
	}
	cfs_critical("connection from bad user %d! - rejected", uid);
	return 1;
}

static void s1_connection_created_fn(
	qb_ipcs_connection_t *c)
{
	struct qb_ipcs_stats srv_stats;

	qb_ipcs_stats_get(s1, &srv_stats, QB_FALSE);

	cfs_debug("Connection created > active:%d > closed:%d",
		    srv_stats.active_connections,
		    srv_stats.closed_connections);
}

static void s1_connection_destroyed_fn(
	qb_ipcs_connection_t *c)
{
	cfs_debug("connection about to be freed");
	
	gpointer ctx;
	if ((ctx = qb_ipcs_context_get(c)))
		g_free(ctx);

}

static int32_t s1_connection_closed_fn(
	qb_ipcs_connection_t *c)
{
	struct qb_ipcs_connection_stats stats;

	qb_ipcs_connection_stats_get(c, &stats, QB_FALSE);

	cfs_debug("Connection to pid:%d destroyed", stats.client_pid);

	return 0;
}

static int32_t s1_msg_process_fn(
	qb_ipcs_connection_t *c,
	void *data,
	size_t size)
{
	struct qb_ipc_request_header *req_pt = 
		(struct qb_ipc_request_header *)data;

	struct s1_context *ctx = (struct s1_context *)qb_ipcs_context_get(c);

	if (!ctx) {
		cfs_critical("qb_ipcs_context_get failed");
		qb_ipcs_disconnect(c);
		return 0;
	}

	int32_t request_id __attribute__ ((aligned(8))) = req_pt->id;
	int32_t request_size __attribute__ ((aligned(8))) = req_pt->size;
	cfs_debug("process msg:%d, size:%d", request_id, request_size);

	char *resp = NULL;

	g_string_truncate(outbuf, 0);

	int32_t result = -ECHRNG;
	if (request_id == CFS_IPC_GET_FS_VERSION) {

		if (request_size != sizeof(struct qb_ipc_request_header)) {
			result = -EINVAL;
		} else {
			result = cfs_create_version_msg(outbuf);
		}

	} else if (request_id == CFS_IPC_GET_CLUSTER_INFO) {

		if (request_size != sizeof(struct qb_ipc_request_header)) {
			result = -EINVAL;
		} else {
			result = cfs_create_memberlist_msg(outbuf);
		}

	} else if (request_id == CFS_IPC_GET_GUEST_LIST) {
		
		if (request_size != sizeof(struct qb_ipc_request_header)) {
			result = -EINVAL;
		} else {
			result = cfs_create_vmlist_msg(outbuf);
		}
	} else if (request_id == CFS_IPC_SET_STATUS) {

		cfs_status_update_request_header_t *rh = 
			(cfs_status_update_request_header_t *)data;

		int datasize = request_size - sizeof(cfs_status_update_request_header_t);

		if (ctx->read_only) {
			result = -EPERM;
		} else if (datasize < 0) {
			result = -EINVAL;
		} else {	
			/* make sure name is 0 terminated */
			rh->name[sizeof(rh->name) - 1] = 0;

			char *dataptr = (char*) data + sizeof(cfs_status_update_request_header_t);

			result = cfs_status_set(rh->name, dataptr, datasize);
		}
	} else if (request_id == CFS_IPC_GET_STATUS) {

		cfs_status_get_request_header_t *rh =
			(cfs_status_get_request_header_t *)data;

		int datasize = request_size - sizeof(cfs_status_get_request_header_t);

		if (datasize < 0) {
			result = -EINVAL;
		} else {	
			/* make sure all names are 0 terminated */
			rh->name[sizeof(rh->name) - 1] = 0;
			rh->nodename[sizeof(rh->nodename) - 1] = 0;

			result = cfs_create_status_msg(outbuf, rh->nodename, rh->name);
		}
	} else if (request_id == CFS_IPC_GET_CONFIG) {

		int pathlen = request_size - sizeof(struct qb_ipc_request_header);

		if (pathlen <= 0) {
			result = -EINVAL;
		} else {
			/* make sure path is 0 terminated */
			((char *)data)[request_size - 1] = 0;
			char *path = (char*) data + sizeof(struct qb_ipc_request_header);

			if (ctx->read_only &&  path_is_private(path)) {
				result = -EPERM;
			} else {
				gpointer tmp = NULL;
				result = memdb_read(memdb, path, &tmp);
				if (result > 0) {
					g_string_append_len(outbuf, tmp, result);
					g_free(tmp);
				}
			}
		}			
	} else if (request_id == CFS_IPC_LOG_CLUSTER_MSG) {

		cfs_log_msg_request_header_t *rh = 
			(cfs_log_msg_request_header_t *)data;

		int datasize = request_size - G_STRUCT_OFFSET(cfs_log_msg_request_header_t, data);
		int msg_len = datasize - rh->ident_len - rh->tag_len;

		if (ctx->read_only) {
			result = -EPERM;
		} else if (msg_len  < 1) {
			result = -EINVAL;
		} else {
			char *msg = rh->data;
			if ((msg[rh->ident_len - 1] == 0) &&
			    (msg[rh->ident_len + rh->tag_len - 1] == 0) &&
			    (((char *)data)[request_size] == 0)) {

				char *ident = msg;
				char *tag = msg + rh->ident_len;
				msg = msg + rh->ident_len + rh->tag_len;

				time_t ctime = time(NULL);
				clog_entry_t *entry = (clog_entry_t *)alloca(CLOG_MAX_ENTRY_SIZE);
				if (clog_pack(entry, cfs.nodename, ident, tag, ctx->client_pid,
					      ctime, rh->priority, msg)) {
					cfs_cluster_log(entry);
				}

				result = 0;

			} else {
				result = -EINVAL;
			}
		}
	} else if (request_id == CFS_IPC_GET_CLUSTER_LOG) {

		cfs_log_get_request_header_t *rh = 
			(cfs_log_get_request_header_t *)data;

		int userlen = request_size - sizeof(cfs_log_get_request_header_t);

		if (userlen <= 0) {
			result = -EINVAL;
		} else {
			/* make sure user string is 0 terminated */
			((char *)data)[request_size - 1] = 0;
			char *user = (char*) data + sizeof(cfs_log_get_request_header_t);

			uint32_t max = rh->max_entries ?  rh->max_entries : 50;
			cfs_cluster_log_dump(outbuf, user, max);
			result = 0;
		}
	} else if (request_id == CFS_IPC_GET_RRD_DUMP) {
	
		if (request_size != sizeof(struct qb_ipc_request_header)) {
			result = -EINVAL;
		} else {
			cfs_rrd_dump(outbuf);
			result = 0;
		}
	} else if (request_id == CFS_IPC_GET_GUEST_CONFIG_PROPERTY) {

		cfs_guest_config_propery_get_request_header_t *rh =
			(cfs_guest_config_propery_get_request_header_t *) data;

		int proplen = request_size - G_STRUCT_OFFSET(cfs_guest_config_propery_get_request_header_t, property);

		result = 0;
		if (rh->vmid < 100 && rh->vmid != 0) {
			cfs_debug("vmid out of range %u", rh->vmid);
			result = -EINVAL;
		} else if (rh->vmid >= 100 && !vmlist_vm_exists(rh->vmid)) {
			result = -ENOENT;
		} else if (proplen <= 0) {
			cfs_debug("proplen <= 0, %d", proplen);
			result = -EINVAL;
		} else {
			((char *)data)[request_size - 1] = 0; // ensure property is 0 terminated

			cfs_debug("cfs_get_guest_config_property: basic valid checked, do request");

			result = cfs_create_guest_conf_property_msg(outbuf, memdb, rh->property, rh->vmid);
		}
	} else if (request_id == CFS_IPC_GET_GUEST_CONFIG_PROPERTIES) {

		cfs_guest_config_properties_get_request_header_t *rh =
			(cfs_guest_config_properties_get_request_header_t *) data;

		size_t remaining = request_size - G_STRUCT_OFFSET(cfs_guest_config_properties_get_request_header_t, props);

		result = 0;
		if (rh->vmid < 100 && rh->vmid != 0) {
			cfs_debug("vmid out of range %u", rh->vmid);
			result = -EINVAL;
		} else if (rh->vmid >= 100 && !vmlist_vm_exists(rh->vmid)) {
			result = -ENOENT;
		} else if (rh->num_props == 0) {
			cfs_debug("num_props == 0");
			result = -EINVAL;
		} else if (remaining <= 1) {
			cfs_debug("property length <= 1, %ld", remaining);
			result = -EINVAL;
		} else {
			const char **properties = malloc(sizeof(char*) * rh->num_props);
			char *current = (rh->props);
			for (uint8_t i = 0; i < rh->num_props; i++) {
			    size_t proplen = strnlen(current, remaining);
			    if (proplen == 0) {
				cfs_debug("property length 0");
				result = -EINVAL;
				break;
			    }
			    if (proplen == remaining || current[proplen] != '\0') {
				cfs_debug("property not \\0 terminated");
				result = -EINVAL;
				break;
			    }
			    if (current[0] < 'a' || current[0] > 'z') {
				cfs_debug("property does not start with [a-z]");
				result = -EINVAL;
				break;
			    }
			    properties[i] = current;
			    remaining -= (proplen + 1);
			    current += proplen + 1;
			}

			if (remaining != 0) {
			    cfs_debug("leftover data after parsing %u properties", rh->num_props);
			    result = -EINVAL;
			}

			if (result == 0) {
			    cfs_debug("cfs_get_guest_config_properties: basic validity checked, do request");
			    result = cfs_create_guest_conf_properties_msg(outbuf, memdb, properties, rh->num_props, rh->vmid);
			}

			free(properties);
		}
	} else if (request_id == CFS_IPC_VERIFY_TOKEN) {

		cfs_verify_token_request_header_t *rh = (cfs_verify_token_request_header_t *) data;
		int tokenlen = request_size - G_STRUCT_OFFSET(cfs_verify_token_request_header_t, token) - 1;

		if (tokenlen <= 0) {
			cfs_debug("tokenlen <= 0, %d", tokenlen);
			result = -EINVAL;
		} else if (memchr(rh->token, '\n', tokenlen) != NULL) {
			cfs_debug("token contains newline");
			result = -EINVAL;
		} else if (rh->token[tokenlen] != '\0') {
			cfs_debug("token not NULL-terminated");
			result = -EINVAL;
		} else if (strnlen(rh->token, tokenlen) != tokenlen) {
			cfs_debug("token contains NULL-byte");
			result = -EINVAL;
		} else {
			cfs_debug("cfs_verify_token: basic validity checked, reading token.cfg");
			gpointer tmp = NULL;
			int bytes_read = memdb_read(memdb, "priv/token.cfg", &tmp);
			size_t remaining = bytes_read > 0 ? bytes_read : 0;
			if (tmp != NULL && remaining >= tokenlen) {
				const char *line = (char *) tmp;
				const char *next_line;
				const char *const end = line + remaining;
				size_t linelen;

				while (line != NULL) {
					next_line = memchr(line, '\n', remaining);
					linelen = next_line == NULL ? remaining : next_line - line;
					if (linelen == tokenlen && strncmp(line, rh->token, linelen) == 0) {
						result = 0;
						break;
					}
					line = next_line;
					if (line != NULL) {
						line += 1;
						remaining = end - line;
					}
				}
				if (line == NULL) {
					result = -ENOENT;
				}
				g_free(tmp);
			} else {
				cfs_debug("token: token.cfg does not exist - ENOENT");
				result = -ENOENT;
			}
		}
	}

	cfs_debug("process result %d", result);

	if (result >= 0) {
		resp = outbuf->str;
		result = 0;
	}

	int iov_len = 2;
	struct iovec iov[iov_len];
	struct qb_ipc_response_header res_header;

	int resp_data_len = resp ? outbuf->len : 0;

	res_header.id = request_id;
	res_header.size = sizeof(res_header) + resp_data_len;
	res_header.error = result;

	iov[0].iov_base = (char *)&res_header;
	iov[0].iov_len = sizeof(res_header);
	iov[1].iov_base = resp;
	iov[1].iov_len = resp_data_len;

	ssize_t res = qb_ipcs_response_sendv(c, iov, iov_len);
	if (res < 0) {
		cfs_critical("qb_ipcs_response_send: %s", strerror(errno));
		qb_ipcs_disconnect(c);
	}

	return 0;
}

static int32_t my_job_add(
	enum qb_loop_priority p, 
	void *data, 
	qb_loop_job_dispatch_fn fn)
{
	return qb_loop_job_add(loop, p, data, fn);
}

static int32_t my_dispatch_add(
	enum qb_loop_priority p, 
	int32_t fd, 
	int32_t evts,
	void *data, 
	qb_ipcs_dispatch_fn_t fn)
{
	return qb_loop_poll_add(loop, p, fd, evts, data, fn);
}

static int32_t my_dispatch_mod(
	enum qb_loop_priority p, 
	int32_t fd, 
	int32_t evts,
	void *data, 
	qb_ipcs_dispatch_fn_t fn)
{
	return qb_loop_poll_mod(loop, p, fd, evts, data, fn);
}

static int32_t my_dispatch_del(
	int32_t fd)
{
	return qb_loop_poll_del(loop, fd);
}

static struct qb_ipcs_service_handlers service_handlers = {
	.connection_accept = s1_connection_accept_fn,
	.connection_created = s1_connection_created_fn,
	.msg_process = s1_msg_process_fn,
	.connection_destroyed = s1_connection_destroyed_fn,
	.connection_closed = s1_connection_closed_fn,
};

static struct qb_ipcs_poll_handlers poll_handlers = {
	.job_add = my_job_add,
	.dispatch_add = my_dispatch_add,
	.dispatch_mod = my_dispatch_mod,
	.dispatch_del = my_dispatch_del,
};

static void timer_job(void *data)
{
	gboolean terminate = FALSE;

	g_mutex_lock (&server_started_mutex);

	if (terminate_server) {
		cfs_debug ("got terminate request");

		if (loop)
			qb_loop_stop (loop);
		
		if (s1) {
			qb_ipcs_destroy (s1);
			s1 = 0;
		}
		server_started = 0;

		g_cond_signal (&server_stopped_cond);
		
		terminate = TRUE;
	} else if (!server_started) {
		server_started = 1;
		g_cond_signal (&server_started_cond);
	}
	
	g_mutex_unlock (&server_started_mutex);

	if (terminate)
		return;
			       
	qb_loop_timer_handle th;
	qb_loop_timer_add(loop, QB_LOOP_LOW, 1000000000, NULL, timer_job, &th);
}

static gpointer worker_thread(gpointer data)
{
	g_return_val_if_fail(loop != NULL, NULL);

	cfs_debug("start event loop");

	qb_ipcs_run(s1);

	qb_loop_timer_handle th;
	qb_loop_timer_add(loop, QB_LOOP_LOW, 1000, NULL, timer_job, &th);

	qb_loop_run(loop);

	cfs_debug("event loop finished - exit worker thread");
	
	return NULL;
}

gboolean server_start(memdb_t *db)
{
	g_return_val_if_fail(loop == NULL, FALSE);
	g_return_val_if_fail(worker == NULL, FALSE);
	g_return_val_if_fail(db != NULL, FALSE);

	terminate_server = 0;
	server_started = 0;
	
	memdb = db;

	outbuf = g_string_sized_new(8192*8);

	if (!(loop = qb_loop_create())) {
		cfs_critical("cant create event loop");
		return FALSE;
	}
	
	s1 = qb_ipcs_create("pve2", 1, QB_IPC_SHM, &service_handlers);
	if (s1 == 0) {
		cfs_critical("qb_ipcs_create failed: %s", strerror(errno));
		return FALSE;
	}
	qb_ipcs_poll_handlers_set(s1, &poll_handlers);

	worker = g_thread_new ("server", worker_thread, NULL);

	g_mutex_lock (&server_started_mutex);
	while (!server_started)
		g_cond_wait (&server_started_cond, &server_started_mutex);
	g_mutex_unlock (&server_started_mutex);
	
	cfs_debug("server started");
	
	return TRUE;
}

void server_stop(void)
{
	cfs_debug("server stop");

	g_mutex_lock (&server_started_mutex);
	terminate_server = 1;
	while (server_started)
		g_cond_wait (&server_stopped_cond, &server_started_mutex);
	g_mutex_unlock (&server_started_mutex);

	if (worker) {
		g_thread_join(worker);
		worker = NULL;
	}
	
	cfs_debug("worker thread finished");

	if (loop) {
		qb_loop_destroy(loop);

		loop = NULL;
	}

	if (outbuf) {
		g_string_free(outbuf, TRUE);
		outbuf = NULL;
	}
}
Commit	Line	Data
fe000966	1	/*
84c98315	2	Copyright (C) 2010 - 2020 Proxmox Server Solutions GmbH
fe000966 DM	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"
95c44445	41	#include "cfs-ipc-ops.h"
fe000966 DM	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;
26754d72 DM	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
fe000966 DM	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
cf1b19d9 TL	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;
fe000966	91
73a31823 DC	92	typedef struct {
	93	struct qb_ipc_request_header req_header;
	94	uint32_t vmid;
	95	uint8_t num_props;
	96	char props[]; /* list of \0 terminated properties */
	97	} cfs_guest_config_properties_get_request_header_t;
	98
7dac5e0e FG	99	typedef struct {
	100	struct qb_ipc_request_header req_header;
	101	char token[];
	102	} cfs_verify_token_request_header_t;
	103
fe000966 DM	104	struct s1_context {
	105	int32_t client_pid;
	106	uid_t uid;
	107	gid_t gid;
	108	gboolean read_only;
	109	};
	110
	111	static int32_t s1_connection_accept_fn(
	112	qb_ipcs_connection_t *c,
	113	uid_t uid,
	114	gid_t gid)
	115	{
	116	if ((uid == 0 && gid == 0) \|\| (gid == cfs.gid)) {
	117	cfs_debug("authenticated connection %d/%d", uid, gid);
	118	struct s1_context *ctx = g_new0(struct s1_context, 1);
	119	ctx->uid = uid;
	120	ctx->gid = gid;
	121	ctx->read_only = (gid == cfs.gid);
	122
	123	struct qb_ipcs_connection_stats stats;
	124	qb_ipcs_connection_stats_get(c, &stats, QB_FALSE);
	125	ctx->client_pid = stats.client_pid;
	126
	127	qb_ipcs_context_set(c, ctx);
	128	return 0;
	129	}
	130	cfs_critical("connection from bad user %d! - rejected", uid);
	131	return 1;
	132	}
	133
	134	static void s1_connection_created_fn(
	135	qb_ipcs_connection_t *c)
	136	{
	137	struct qb_ipcs_stats srv_stats;
	138
	139	qb_ipcs_stats_get(s1, &srv_stats, QB_FALSE);
	140
	141	cfs_debug("Connection created > active:%d > closed:%d",
	142	srv_stats.active_connections,
	143	srv_stats.closed_connections);
	144	}
	145
	146	static void s1_connection_destroyed_fn(
	147	qb_ipcs_connection_t *c)
	148	{
	149	cfs_debug("connection about to be freed");
	150
	151	gpointer ctx;
	152	if ((ctx = qb_ipcs_context_get(c)))
	153	g_free(ctx);
	154
	155	}
	156
	157	static int32_t s1_connection_closed_fn(
	158	qb_ipcs_connection_t *c)
	159	{
	160	struct qb_ipcs_connection_stats stats;
	161
	162	qb_ipcs_connection_stats_get(c, &stats, QB_FALSE);
	163
	164	cfs_debug("Connection to pid:%d destroyed", stats.client_pid);
	165
	166	return 0;
	167	}
168
169	static int32_t s1_msg_process_fn(
170	qb_ipcs_connection_t *c,
bd087c23	171	void *data,
fe000966 DM	172	size_t size)
	173	{
	174	struct qb_ipc_request_header *req_pt =
	175	(struct qb_ipc_request_header *)data;
	176
	177	struct s1_context ctx = (struct s1_context )qb_ipcs_context_get(c);
	178
	179	if (!ctx) {
	180	cfs_critical("qb_ipcs_context_get failed");
	181	qb_ipcs_disconnect(c);
	182	return 0;
	183	}
	184
8dbade24 TL	185	int32_t request_id __attribute__ ((aligned(8))) = req_pt->id;
	186	int32_t request_size __attribute__ ((aligned(8))) = req_pt->size;
	187	cfs_debug("process msg:%d, size:%d", request_id, request_size);
fe000966 DM	188
	189	char *resp = NULL;
	190
	191	g_string_truncate(outbuf, 0);
	192
	193	int32_t result = -ECHRNG;
8dbade24	194	if (request_id == CFS_IPC_GET_FS_VERSION) {
fe000966	195
8dbade24	196	if (request_size != sizeof(struct qb_ipc_request_header)) {
fe000966 DM	197	result = -EINVAL;
	198	} else {
	199	result = cfs_create_version_msg(outbuf);
	200	}
	201
8dbade24	202	} else if (request_id == CFS_IPC_GET_CLUSTER_INFO) {
fe000966	203
8dbade24	204	if (request_size != sizeof(struct qb_ipc_request_header)) {
fe000966 DM	205	result = -EINVAL;
	206	} else {
	207	result = cfs_create_memberlist_msg(outbuf);
	208	}
	209
8dbade24	210	} else if (request_id == CFS_IPC_GET_GUEST_LIST) {
fe000966	211
8dbade24	212	if (request_size != sizeof(struct qb_ipc_request_header)) {
fe000966 DM	213	result = -EINVAL;
	214	} else {
	215	result = cfs_create_vmlist_msg(outbuf);
	216	}
8dbade24	217	} else if (request_id == CFS_IPC_SET_STATUS) {
fe000966 DM	218
	219	cfs_status_update_request_header_t *rh =
	220	(cfs_status_update_request_header_t *)data;
	221
8dbade24	222	int datasize = request_size - sizeof(cfs_status_update_request_header_t);
fe000966 DM	223
	224	if (ctx->read_only) {
	225	result = -EPERM;
	226	} else if (datasize < 0) {
	227	result = -EINVAL;
	228	} else {
	229	/* make sure name is 0 terminated */
	230	rh->name[sizeof(rh->name) - 1] = 0;
	231
bd087c23	232	char dataptr = (char) data + sizeof(cfs_status_update_request_header_t);
fe000966 DM	233
	234	result = cfs_status_set(rh->name, dataptr, datasize);
	235	}
8dbade24	236	} else if (request_id == CFS_IPC_GET_STATUS) {
fe000966 DM	237
	238	cfs_status_get_request_header_t *rh =
	239	(cfs_status_get_request_header_t *)data;
	240
8dbade24	241	int datasize = request_size - sizeof(cfs_status_get_request_header_t);
fe000966 DM	242
	243	if (datasize < 0) {
	244	result = -EINVAL;
	245	} else {
	246	/* make sure all names are 0 terminated */
	247	rh->name[sizeof(rh->name) - 1] = 0;
	248	rh->nodename[sizeof(rh->nodename) - 1] = 0;
	249
	250	result = cfs_create_status_msg(outbuf, rh->nodename, rh->name);
	251	}
8dbade24	252	} else if (request_id == CFS_IPC_GET_CONFIG) {
fe000966	253
8dbade24	254	int pathlen = request_size - sizeof(struct qb_ipc_request_header);
fe000966 DM	255
	256	if (pathlen <= 0) {
	257	result = -EINVAL;
	258	} else {
	259	/* make sure path is 0 terminated */
78c6e1de	260	((char *)data)[request_size - 1] = 0;
bd087c23	261	char path = (char) data + sizeof(struct qb_ipc_request_header);
fe000966 DM	262
	263	if (ctx->read_only && path_is_private(path)) {
	264	result = -EPERM;
	265	} else {
	266	gpointer tmp = NULL;
	267	result = memdb_read(memdb, path, &tmp);
4a9c52e8	268	if (result > 0) {
fe000966 DM	269	g_string_append_len(outbuf, tmp, result);
	270	g_free(tmp);
	271	}
	272	}
	273	}
8dbade24	274	} else if (request_id == CFS_IPC_LOG_CLUSTER_MSG) {
fe000966 DM	275
	276	cfs_log_msg_request_header_t *rh =
	277	(cfs_log_msg_request_header_t *)data;
	278
8dbade24	279	int datasize = request_size - G_STRUCT_OFFSET(cfs_log_msg_request_header_t, data);
fe000966 DM	280	int msg_len = datasize - rh->ident_len - rh->tag_len;
	281
	282	if (ctx->read_only) {
	283	result = -EPERM;
	284	} else if (msg_len < 1) {
	285	result = -EINVAL;
	286	} else {
	287	char *msg = rh->data;
	288	if ((msg[rh->ident_len - 1] == 0) &&
	289	(msg[rh->ident_len + rh->tag_len - 1] == 0) &&
8dbade24	290	(((char *)data)[request_size] == 0)) {
fe000966 DM	291
	292	char *ident = msg;
	293	char *tag = msg + rh->ident_len;
	294	msg = msg + rh->ident_len + rh->tag_len;
	295
	296	time_t ctime = time(NULL);
	297	clog_entry_t entry = (clog_entry_t )alloca(CLOG_MAX_ENTRY_SIZE);
	298	if (clog_pack(entry, cfs.nodename, ident, tag, ctx->client_pid,
	299	ctime, rh->priority, msg)) {
	300	cfs_cluster_log(entry);
	301	}
	302
	303	result = 0;
	304
	305	} else {
	306	result = -EINVAL;
	307	}
	308	}
8dbade24	309	} else if (request_id == CFS_IPC_GET_CLUSTER_LOG) {
fe000966 DM	310
	311	cfs_log_get_request_header_t *rh =
	312	(cfs_log_get_request_header_t *)data;
	313
8dbade24	314	int userlen = request_size - sizeof(cfs_log_get_request_header_t);
fe000966 DM	315
	316	if (userlen <= 0) {
	317	result = -EINVAL;
	318	} else {
	319	/* make sure user string is 0 terminated */
78c6e1de	320	((char *)data)[request_size - 1] = 0;
bd087c23	321	char user = (char) data + sizeof(cfs_log_get_request_header_t);
fe000966 DM	322
	323	uint32_t max = rh->max_entries ? rh->max_entries : 50;
	324	cfs_cluster_log_dump(outbuf, user, max);
	325	result = 0;
	326	}
8dbade24	327	} else if (request_id == CFS_IPC_GET_RRD_DUMP) {
fe000966	328
8dbade24	329	if (request_size != sizeof(struct qb_ipc_request_header)) {
fe000966 DM	330	result = -EINVAL;
	331	} else {
	332	cfs_rrd_dump(outbuf);
	333	result = 0;
	334	}
8dbade24	335	} else if (request_id == CFS_IPC_GET_GUEST_CONFIG_PROPERTY) {
cf1b19d9 TL	336
	337	cfs_guest_config_propery_get_request_header_t *rh =
	338	(cfs_guest_config_propery_get_request_header_t *) data;
	339
8dbade24	340	int proplen = request_size - G_STRUCT_OFFSET(cfs_guest_config_propery_get_request_header_t, property);
cf1b19d9 TL	341
	342	result = 0;
	343	if (rh->vmid < 100 && rh->vmid != 0) {
	344	cfs_debug("vmid out of range %u", rh->vmid);
	345	result = -EINVAL;
	346	} else if (rh->vmid >= 100 && !vmlist_vm_exists(rh->vmid)) {
	347	result = -ENOENT;
	348	} else if (proplen <= 0) {
	349	cfs_debug("proplen <= 0, %d", proplen);
	350	result = -EINVAL;
	351	} else {
78c6e1de	352	((char *)data)[request_size - 1] = 0; // ensure property is 0 terminated
cf1b19d9 TL	353
	354	cfs_debug("cfs_get_guest_config_property: basic valid checked, do request");
	355
	356	result = cfs_create_guest_conf_property_msg(outbuf, memdb, rh->property, rh->vmid);
	357	}
73a31823 DC	358	} else if (request_id == CFS_IPC_GET_GUEST_CONFIG_PROPERTIES) {
	359
	360	cfs_guest_config_properties_get_request_header_t *rh =
	361	(cfs_guest_config_properties_get_request_header_t *) data;
	362
	363	size_t remaining = request_size - G_STRUCT_OFFSET(cfs_guest_config_properties_get_request_header_t, props);
	364
	365	result = 0;
	366	if (rh->vmid < 100 && rh->vmid != 0) {
	367	cfs_debug("vmid out of range %u", rh->vmid);
	368	result = -EINVAL;
	369	} else if (rh->vmid >= 100 && !vmlist_vm_exists(rh->vmid)) {
	370	result = -ENOENT;
	371	} else if (rh->num_props == 0) {
	372	cfs_debug("num_props == 0");
	373	result = -EINVAL;
	374	} else if (remaining <= 1) {
	375	cfs_debug("property length <= 1, %ld", remaining);
	376	result = -EINVAL;
	377	} else {
	378	const char *properties = malloc(sizeof(char) * rh->num_props);
	379	char *current = (rh->props);
	380	for (uint8_t i = 0; i < rh->num_props; i++) {
	381	size_t proplen = strnlen(current, remaining);
	382	if (proplen == 0) {
	383	cfs_debug("property length 0");
	384	result = -EINVAL;
	385	break;
	386	}
	387	if (proplen == remaining \|\| current[proplen] != '\0') {
	388	cfs_debug("property not \\0 terminated");
	389	result = -EINVAL;
	390	break;
	391	}
	392	if (current[0] < 'a' \|\| current[0] > 'z') {
	393	cfs_debug("property does not start with [a-z]");
	394	result = -EINVAL;
	395	break;
	396	}
	397	properties[i] = current;
	398	remaining -= (proplen + 1);
	399	current += proplen + 1;
	400	}
	401
	402	if (remaining != 0) {
	403	cfs_debug("leftover data after parsing %u properties", rh->num_props);
	404	result = -EINVAL;
	405	}
	406
	407	if (result == 0) {
	408	cfs_debug("cfs_get_guest_config_properties: basic validity checked, do request");
	409	result = cfs_create_guest_conf_properties_msg(outbuf, memdb, properties, rh->num_props, rh->vmid);
	410	}
	411
	412	free(properties);
	413	}
7dac5e0e FG	414	} else if (request_id == CFS_IPC_VERIFY_TOKEN) {
	415
	416	cfs_verify_token_request_header_t rh = (cfs_verify_token_request_header_t ) data;
	417	int tokenlen = request_size - G_STRUCT_OFFSET(cfs_verify_token_request_header_t, token) - 1;
	418
	419	if (tokenlen <= 0) {
	420	cfs_debug("tokenlen <= 0, %d", tokenlen);
	421	result = -EINVAL;
	422	} else if (memchr(rh->token, '\n', tokenlen) != NULL) {
	423	cfs_debug("token contains newline");
	424	result = -EINVAL;
	425	} else if (rh->token[tokenlen] != '\0') {
	426	cfs_debug("token not NULL-terminated");
	427	result = -EINVAL;
	428	} else if (strnlen(rh->token, tokenlen) != tokenlen) {
	429	cfs_debug("token contains NULL-byte");
	430	result = -EINVAL;
	431	} else {
	432	cfs_debug("cfs_verify_token: basic validity checked, reading token.cfg");
	433	gpointer tmp = NULL;
	434	int bytes_read = memdb_read(memdb, "priv/token.cfg", &tmp);
	435	size_t remaining = bytes_read > 0 ? bytes_read : 0;
	436	if (tmp != NULL && remaining >= tokenlen) {
92f54a87 TL	437	const char line = (char ) tmp;
92f54a87 TL	438	const char *next_line;
7dac5e0e FG	439	const char *const end = line + remaining;
	440	size_t linelen;
	441
	442	while (line != NULL) {
	443	next_line = memchr(line, '\n', remaining);
	444	linelen = next_line == NULL ? remaining : next_line - line;
	445	if (linelen == tokenlen && strncmp(line, rh->token, linelen) == 0) {
	446	result = 0;
	447	break;
	448	}
	449	line = next_line;
	450	if (line != NULL) {
	451	line += 1;
	452	remaining = end - line;
	453	}
	454	}
	455	if (line == NULL) {
	456	result = -ENOENT;
	457	}
	458	g_free(tmp);
	459	} else {
	460	cfs_debug("token: token.cfg does not exist - ENOENT");
	461	result = -ENOENT;
	462	}
	463	}
fe000966 DM	464	}
	465
	466	cfs_debug("process result %d", result);
	467
	468	if (result >= 0) {
	469	resp = outbuf->str;
	470	result = 0;
	471	}
	472
	473	int iov_len = 2;
	474	struct iovec iov[iov_len];
	475	struct qb_ipc_response_header res_header;
	476
	477	int resp_data_len = resp ? outbuf->len : 0;
	478
8dbade24	479	res_header.id = request_id;
fe000966 DM	480	res_header.size = sizeof(res_header) + resp_data_len;
	481	res_header.error = result;
	482
	483	iov[0].iov_base = (char *)&res_header;
	484	iov[0].iov_len = sizeof(res_header);
	485	iov[1].iov_base = resp;
	486	iov[1].iov_len = resp_data_len;
	487
	488	ssize_t res = qb_ipcs_response_sendv(c, iov, iov_len);
	489	if (res < 0) {
	490	cfs_critical("qb_ipcs_response_send: %s", strerror(errno));
	491	qb_ipcs_disconnect(c);
	492	}
	493
	494	return 0;
	495	}
	496
	497	static int32_t my_job_add(
	498	enum qb_loop_priority p,
	499	void *data,
	500	qb_loop_job_dispatch_fn fn)
	501	{
	502	return qb_loop_job_add(loop, p, data, fn);
	503	}
	504
	505	static int32_t my_dispatch_add(
	506	enum qb_loop_priority p,
	507	int32_t fd,
	508	int32_t evts,
	509	void *data,
	510	qb_ipcs_dispatch_fn_t fn)
	511	{
	512	return qb_loop_poll_add(loop, p, fd, evts, data, fn);
	513	}
	514
	515	static int32_t my_dispatch_mod(
	516	enum qb_loop_priority p,
	517	int32_t fd,
	518	int32_t evts,
	519	void *data,
	520	qb_ipcs_dispatch_fn_t fn)
	521	{
	522	return qb_loop_poll_mod(loop, p, fd, evts, data, fn);
	523	}
	524
	525	static int32_t my_dispatch_del(
	526	int32_t fd)
	527	{
	528	return qb_loop_poll_del(loop, fd);
	529	}
	530
	531	static struct qb_ipcs_service_handlers service_handlers = {
	532	.connection_accept = s1_connection_accept_fn,
	533	.connection_created = s1_connection_created_fn,
	534	.msg_process = s1_msg_process_fn,
	535	.connection_destroyed = s1_connection_destroyed_fn,
	536	.connection_closed = s1_connection_closed_fn,
	537	};
	538
	539	static struct qb_ipcs_poll_handlers poll_handlers = {
	540	.job_add = my_job_add,
	541	.dispatch_add = my_dispatch_add,
	542	.dispatch_mod = my_dispatch_mod,
	543	.dispatch_del = my_dispatch_del,
544	};
545
546	static void timer_job(void *data)
547	{
26754d72 DM	548	gboolean terminate = FALSE;
	549
	550	g_mutex_lock (&server_started_mutex);
	551
	552	if (terminate_server) {
	553	cfs_debug ("got terminate request");
	554
	555	if (loop)
	556	qb_loop_stop (loop);
	557
	558	if (s1) {
	559	qb_ipcs_destroy (s1);
	560	s1 = 0;
	561	}
	562	server_started = 0;
	563
	564	g_cond_signal (&server_stopped_cond);
	565
	566	terminate = TRUE;
	567	} else if (!server_started) {
	568	server_started = 1;
	569	g_cond_signal (&server_started_cond);
	570	}
	571
	572	g_mutex_unlock (&server_started_mutex);
fe000966	573
26754d72 DM	574	if (terminate)
	575	return;
	576
fe000966 DM	577	qb_loop_timer_handle th;
	578	qb_loop_timer_add(loop, QB_LOOP_LOW, 1000000000, NULL, timer_job, &th);
	579	}
	580
	581	static gpointer worker_thread(gpointer data)
	582	{
	583	g_return_val_if_fail(loop != NULL, NULL);
	584
	585	cfs_debug("start event loop");
	586
	587	qb_ipcs_run(s1);
	588
	589	qb_loop_timer_handle th;
	590	qb_loop_timer_add(loop, QB_LOOP_LOW, 1000, NULL, timer_job, &th);
	591
	592	qb_loop_run(loop);
	593
	594	cfs_debug("event loop finished - exit worker thread");
	595
	596	return NULL;
	597	}
	598
	599	gboolean server_start(memdb_t *db)
	600	{
	601	g_return_val_if_fail(loop == NULL, FALSE);
	602	g_return_val_if_fail(worker == NULL, FALSE);
	603	g_return_val_if_fail(db != NULL, FALSE);
	604
26754d72 DM	605	terminate_server = 0;
	606	server_started = 0;
	607
fe000966 DM	608	memdb = db;
	609
	610	outbuf = g_string_sized_new(8192*8);
	611
	612	if (!(loop = qb_loop_create())) {
	613	cfs_critical("cant create event loop");
	614	return FALSE;
	615	}
	616
	617	s1 = qb_ipcs_create("pve2", 1, QB_IPC_SHM, &service_handlers);
	618	if (s1 == 0) {
	619	cfs_critical("qb_ipcs_create failed: %s", strerror(errno));
	620	return FALSE;
	621	}
	622	qb_ipcs_poll_handlers_set(s1, &poll_handlers);
26754d72	623
89fde9ac	624	worker = g_thread_new ("server", worker_thread, NULL);
fe000966	625
26754d72 DM	626	g_mutex_lock (&server_started_mutex);
	627	while (!server_started)
	628	g_cond_wait (&server_started_cond, &server_started_mutex);
	629	g_mutex_unlock (&server_started_mutex);
	630
	631	cfs_debug("server started");
	632
fe000966 DM	633	return TRUE;
	634	}
	635
	636	void server_stop(void)
	637	{
	638	cfs_debug("server stop");
	639
26754d72 DM	640	g_mutex_lock (&server_started_mutex);
	641	terminate_server = 1;
	642	while (server_started)
	643	g_cond_wait (&server_stopped_cond, &server_started_mutex);
	644	g_mutex_unlock (&server_started_mutex);
fe000966 DM	645
	646	if (worker) {
	647	g_thread_join(worker);
	648	worker = NULL;
	649	}
	650
	651	cfs_debug("worker thread finished");
	652
	653	if (loop) {
	654	qb_loop_destroy(loop);
	655
	656	loop = NULL;
	657	}
	658
	659	if (outbuf) {
	660	g_string_free(outbuf, TRUE);
	661	outbuf = NULL;
	662	}
	663	}