rmps/rmps.c

#include "log.h"
#include "confparser.h"
#include "agent_pool.h"
#include "client_pool.h"
#include "job_queue.h"
#include "rmps.h"
#include <pthread.h>
#include <string.h>
#include <unistd.h>
#include <sys/stat.h>
#include <fcntl.h>

/* included for openssl */
#include <openssl/ssl.h>
#include <openssl/err.h>

static void rmps_shutdown(void);
static void signal_handler(int sig);
static void set_env(void);
static void daemonize(const char *rundir);
static void spawn_pidfile(const char *pidfile);
static inline int set_reuse_addr(int sockfd);
static int open_listener(int port);
static void cleanup(void);
static void signal_handler(int sig);
static void load_certificates(SSL_CTX *ctx, const char *certfile,
	const char *keyfile, const char *cafile);
static SSL_CTX *init_server_ctx(const char *cipherlist, int mode);

static int pid_file_handle;

static void cleanup(void)
{
	log(VERBOSE, "Deleting pidfile %s", conf.rmps.pidfile);
	if (unlink(conf.rmps.pidfile) != 0)
		log(WARNING,
			"Failed to delete pidfile %s. Reason code: %d",
			conf.rmps.pidfile, errno);
}

static void signal_handler(int sig)
{
	switch (sig) {
	case SIGHUP:
		log(WARNING, "Received SIGHUP signal. Ignoring...");
		break;
	case SIGINT:
	case SIGTERM:
		log(INFO, "Received SIGTERM signal.");
		log(INFO, "RMPS is shutting down...");
		rmps_shutdown();
		log(INFO, "RMPS has been stopped properly.");
		_exit(EXIT_SUCCESS);
		break;
	default:
		log(WARNING, "Unhandled signal %s", strsignal(sig));
		break;
	}
}

static void rmps_shutdown(void)
{
	close(pid_file_handle);
	cleanup();
}

static void set_env(void)
{
	struct sigaction new_sigaction;
	sigset_t new_sigset;

	/* Set signal mask - signals we want to block */
	sigemptyset(&new_sigset);
	sigaddset(&new_sigset, SIGCHLD);  /* ignore child */
	sigaddset(&new_sigset, SIGTSTP);  /* ignore Tty stop signals */
	sigaddset(&new_sigset, SIGTTOU);  /* ignore Tty background writes */
	sigaddset(&new_sigset, SIGTTIN);  /* ignore Tty background reads */
	sigprocmask(SIG_BLOCK, &new_sigset, NULL); /* Block above signals */

	/* Set up a signal handler */
	new_sigaction.sa_handler = signal_handler;
	sigemptyset(&new_sigaction.sa_mask);
	new_sigaction.sa_flags = 0;

	/* Signals to handle */
	sigaction(SIGHUP, &new_sigaction, NULL);  /* catch hangup signal */
	sigaction(SIGTERM, &new_sigaction, NULL); /* catch term signal */
	sigaction(SIGINT, &new_sigaction, NULL);  /* catch interrupt signal */
	signal(SIGPIPE, SIG_IGN); /* prevent crashing from bad writes */
}

static void daemonize(const char *rundir)
{
	int pid, sid, i;
	/* Check if parent process id is set */
	if (getppid() == 1) {
		/* PPID exists, therefore we are already a daemon */
		return;
	}
	/* Fork*/
	pid = fork();
	if (pid < 0) {
		/* Could not fork */
		log(ERROR, "Failed to fork the parent process. Exiting!");
		exit(EXIT_FAILURE);
	}
	if (pid > 0) {
		/* Child created ok, so exit parent process */
		log(INFO, "Child process created: %d", pid);
		exit(EXIT_SUCCESS);
	}
	/* Child continues */
	umask(027); /* Set file permissions 750 */
	/* Get a new process group */
	sid = setsid();
	if (sid < 0) {
		log(ERROR, "Failed to create a process group. Exiting!");
		exit(EXIT_FAILURE);
	}
	/* Close all file descriptors because we fork */
	close(0); /* stdin  */
	close(1); /* stdout */
	close(2); /* stderr */
	/* Route I/O connections */
	/* Open STDIN */
	i = open("/dev/null", O_RDWR);
	/* STDOUT */
	dup(i);
	/* STDERR */
	dup(i);
	chdir(rundir); /* change running directory */
}

static void spawn_pidfile(const char *pidfile)
{
	char str[10];
	/* Ensure only one copy */
	pid_file_handle = open(pidfile, O_RDWR|O_CREAT, 0600);
	if (pid_file_handle == -1) {
		/* Couldn't open lock file */
		log(ERROR, "Could not create PID file %s - Exiting!", pidfile);
		exit(EXIT_FAILURE);
	}

	/* Try to lock file */
	if (lockf(pid_file_handle, F_TLOCK, 0) == -1) {
		/* Couldn't get lock on lock file */
		log(ERROR, "Could not lock PID file %s - Exiting!", pidfile);
		exit(EXIT_FAILURE);
	}

	/* Get and format PID */
	sprintf(str, "%d\n", getpid());
	/* write pid to lockfile */
	write(pid_file_handle, str, strlen(str));
}

static inline int set_reuse_addr(int sockfd)
{
	int yes = 1;

	return setsockopt(sockfd, SOL_SOCKET, SO_REUSEADDR,
		&yes, sizeof(yes));
}

static int open_listener(int port)
{
	int sd;
	struct sockaddr_in addr;

	bzero(&addr, sizeof(addr));
	addr.sin_family = AF_INET;
	addr.sin_port = htons(port);
	addr.sin_addr.s_addr = INADDR_ANY;
	sd = socket(addr.sin_family, SOCK_STREAM, 0);
	if (sd < 0) {
		log(ERROR, "Failed to create socket");
		goto exit;
	}
	if (set_reuse_addr(sd) < 0) {
		log(ERROR,
			"Failed to set reuse on address - Aborting...", port);
		goto exit;
	}
	if (bind(sd, (struct sockaddr *)&addr, sizeof(addr)) != 0) {
		log(ERROR, "Failed to bind on port: %d - Aborting...", port);
		goto exit;
	}
	if (listen(sd, 10) != 0) {
		log(ERROR,
			"Failed to start listener on port %d - Aborting...",
			port);
		goto exit;
	}
	return sd;
exit:
	log(INFO, "RMPS failed to start, shutting down...");
	atexit(cleanup);
}

/* Init server and create context */
static SSL_CTX *init_server_ctx(const char *cipherlist, int mode)
{
	SSL_CTX *ctx;
	char ciphers[1024];

//	OpenSSL_add_all_algorithms();	/* load & register all cryptos, etc. */
//	OpenSSL_add_all_ciphers();	/* load & register all cryptos, etc. */
	SSL_load_error_strings();	/* load all error messages */
	SSL_library_init();

	/* create new context from method */
	ctx = SSL_CTX_new(TLS_method());
	if (ctx == NULL) {
		log(ERROR, "SSL_CTX_new() returned NULL - Aborting...");
		log(ERROR, "RMPS failed to start, shutting down...");
		exit(EXIT_FAILURE);
	}
	SSL_CTX_set_verify(ctx, mode, NULL);

	ciphers[0] = 0;
	strcat(ciphers, "-ALL"); /* Disable any ciphers we have by default */
	strcat(ciphers, cipherlist);
	/* This is very delicate, try to understand the ciphers */
	SSL_CTX_set_cipher_list(ctx, cipherlist);
	log(VERBOSE, "cipherlist = %s", cipherlist);

	return ctx;
}

/*-------------------------------------------*/
/*--- LoadCertificates - load from files. ---*/
/*-------------------------------------------*/
void load_certificates(SSL_CTX *ctx, const char *certfile,
		const char *keyfile, const char *cafile)
{
	/* set the local certificate from certfile */
	if (SSL_CTX_use_certificate_file(ctx, certfile, SSL_FILETYPE_PEM) <= 0) {
		log(ERROR, "Failed to load certfile! SSL error below:");
		log_ssl();
		log(INFO, "RMPS failed to start, shutting down...");
		atexit(cleanup);
	}
	/* set the private key from KeyFile (may be the same as CertFile) */
	if (SSL_CTX_use_PrivateKey_file(ctx, keyfile, SSL_FILETYPE_PEM) <= 0) {
		log(ERROR, "Failed to load keyfile! SSL error below:");
		log_ssl();
		log(INFO, "RMPS failed to start, shutting down...");
		atexit(cleanup);
	}
	/* verify private key */
	if (!SSL_CTX_check_private_key(ctx)) {
		log(ERROR,
			"Private key does not match the public certificate.");
		log(INFO, "RMPS failed to start, shutting down...");
		atexit(cleanup);
	}
	if (cafile != NULL) {
		SSL_CTX_set_client_CA_list(ctx,
			SSL_load_client_CA_file(cafile));
		SSL_CTX_load_verify_locations(ctx, cafile, NULL);
		//SSL_CTX_set_verify_depth(ctx, 1);
	}
}

void launch_rmps(struct conf_table *conf, int fork_flag)
{
	pthread_t pool[2];
	struct pool_data pool_args[2];

	log(INFO, "Starting up RMPS...");
	/* Set signal handling */
	set_env();
	/* Deamonize */
	if (fork_flag)
		daemonize("/tmp/");
	/* Spawn & lock pidfile */
	spawn_pidfile(conf->rmps.pidfile);

	/* openssl req -x509 -newkey rsa:2048 -keyout key.pem -out cert.pem -days XXX -nodes
	 * -nodes is for not protecing with a passphrase
	 * http://stackoverflow.com/questions/10175812/how-to-create-a-self-signed-certificate-with-openssl
	 */
	pool_args[0].ctx = init_server_ctx(conf->rmps.cipherlist,
			SSL_VERIFY_PEER |
			SSL_VERIFY_CLIENT_ONCE |
			SSL_VERIFY_FAIL_IF_NO_PEER_CERT);
	log(VERBOSE, "Loading agent certs and keys.");
	load_certificates(pool_args[0].ctx, conf->rmps.agent_tls_crt,
		conf->rmps.agent_tls_key, conf->rmps.cafile);
	log(VERBOSE, "Starting agent listener on port: %d",
		atoi(conf->rmps.agent_port));
	pool_args[0].srv = open_listener(atoi(conf->rmps.agent_port));
	pool_args[0].size = conf->rmps.agent_poolsize;
	log(VERBOSE, "Creating agent thread pool (mutex).");
	pthread_create(&pool[0], NULL, agent_pool, &pool_args[0]);

	pool_args[1].ctx = init_server_ctx(conf->rmps.cipherlist,
		SSL_VERIFY_NONE);
	log(VERBOSE, "Loading client certs and keys.");
	load_certificates(pool_args[1].ctx, conf->rmps.client_tls_crt,
		conf->rmps.client_tls_key, conf->rmps.cafile);
	log(VERBOSE, "Starting client listener on port: %d",
		atoi(conf->rmps.client_port));
	pool_args[1].srv = open_listener(atoi(conf->rmps.client_port));
	pool_args[1].size = conf->rmps.client_poolsize;
	log(VERBOSE, "Creating client thread pool (mutex).");
	pthread_create(&pool[1], NULL, client_pool, &pool_args[1]);
	if (start_job_queue(conf->rmps.agent_poolsize) == FAIL) {
		log(ERROR, "On start_job_queue(), RMPS failed to start, shutting down...");
		atexit(cleanup);
		return;
	}
	pthread_join(pool[0], NULL);
	pthread_join(pool[1], NULL);
}