rmps/agent/agent.c

#include <fcntl.h>
#include <unistd.h>
#include <pthread.h>
#include <string.h>
#include "job.h"
#include "../protocol.h"
#include "agent_ssl.h"

#define FAIL -1

static struct job_args *args;
static pthread_t *job_thread;

static short get_job_slot();

static short get_job_slot()
{
	unsigned short i;
	for (i = 0; i < MAX_AGENT_JOBS; i++)
		if (args[i].slot != FULL)
			return i;
	return -1;
}

static short find_job(unsigned id)
{
	unsigned short i;
	for (i = 0; i < MAX_AGENT_JOBS; i++)
		if (args[i].buf.meta.id == id)
			return i;
	return -1;
}

int main(int count, char *strings[])
{   
	SSL_CTX *ctx;
	int server;
	SSL *ssl;
	int bytes, i;
	char *hostname, *portnum;
    
	if (count != 6) {
		printf("usage: %s <hostname> <portnum> <key> <cert> <CA>\n", strings[0]);
		_exit(EXIT_SUCCESS);
	}
	hostname=strings[1];
	portnum=strings[2];
	if ((ctx = init_ctx()) == NULL)
		_exit(EXIT_FAILURE);
	server = connect_to_rmps(hostname, atoi(portnum));
	if (!server) {
		fprintf(stderr, "Failed to connect to RMPS: %s:%d\n", hostname, atoi(portnum));
		_exit(EXIT_FAILURE);
	}
	load_certs(ctx, strings[3], strings[4], strings[5]);
	ssl = SSL_new(ctx);		/* create new SSL connection state */
	SSL_set_fd(ssl, server);	/* attach the socket descriptor */
    
	if (SSL_connect(ssl) == FAIL) {	/* perform the connection */
		ERR_print_errors_fp(stderr);
		close(server);
		SSL_CTX_free(ctx);
		_exit(EXIT_FAILURE);
	}
	printf("Connected with %s encryption\n", SSL_get_cipher(ssl));
	show_certs(ssl);
	if (!(args = (struct job_args*)calloc(1, sizeof(struct job_args) * MAX_AGENT_JOBS))) {
		fprintf( stderr,
			 "Failed to calloc() %d bytes for job_args! Exiting...\n",
			 (int)sizeof(struct job_args) * MAX_AGENT_JOBS );
		SSL_shutdown(ssl);
		SSL_free(ssl);
		close(server);
		SSL_CTX_free(ctx);
		_exit(EXIT_FAILURE);
	}
	if (!(job_thread = (pthread_t*)calloc(1, sizeof(pthread_t) * MAX_AGENT_JOBS))) {
		fprintf( stderr,
			 "Failed to calloc() %d bytes for job_threads! Exiting...\n",
			 (int)sizeof(pthread_t) * MAX_AGENT_JOBS );
		SSL_shutdown(ssl);
		SSL_free(ssl);
		close(server);
		SSL_CTX_free(ctx);
		free(args);
		_exit(EXIT_FAILURE);
	}
	for (i = 0; i < MAX_AGENT_JOBS; i++) {
		args[i].slot = FREE;
		args[i].ssl = ssl;
	}

	do {
		struct msg_t buf;
		memset(&buf, 0, sizeof(struct msg_t));
		bytes = SSL_read(ssl, &buf, sizeof(struct msg_t));
		if (bytes > 0) {
			short index;
			if (bytes != sizeof(struct msg_t)) {
				fprintf( stderr,
					 "Received non-standard data from server!\n" );
				continue;
			}
			if (buf.chunk.id == 0) {
				if ((index = get_job_slot()) == FAIL) {
					buf.chunk.id = -1; /* ID -1 means reject (full) */
					sprintf((char*)buf.chunk.data, "The agent's queue is full!");
					SSL_write(ssl, &buf, sizeof(struct msg_t));
					continue;
				}
				args[index].slot = FULL;
				memcpy(&args[index].buf, &buf, sizeof(struct msg_t));
				switch (args[index].buf.meta.type) {
				case UNIX:
					pthread_create( &job_thread[index],
							NULL,
							exec_unix,
							&args[index] );
					continue;
				case INSTALL_PKG:
					pthread_create( &job_thread[index],
							NULL,
							install_pkg,
							&args[index] );
					continue;
				case QUERY_PKG:
					pthread_create( &job_thread[index],
							NULL,
							query_pkg,
							&args[index] );
					continue;
				case DELETE_PKG:
					pthread_create( &job_thread[index],
							NULL,
							delete_pkg,
							&args[index] );
					continue;
				case LIST_PKGS:
					pthread_create( &job_thread[index],
							NULL,
							list_pkgs,
							&args[index] );
					continue;
				case UPDATE_PKG:
					pthread_create( &job_thread[index],
							NULL,
							update_pkg,
							&args[index] );
					continue;
				case UPDATE_PKGS:
					pthread_create( &job_thread[index],
							NULL,
							update_pkgs,
							&args[index] );
					continue;
				case GET_OS:
					pthread_create( &job_thread[index],
							NULL,
							get_os,
							&args[index] );
					continue;
				case GET_KERNEL:
					pthread_create( &job_thread[index],
							NULL,
							get_kernel,
							&args[index] );
					continue;
				case GET_UPTIME:
					pthread_create( &job_thread[index],
							NULL,
							get_uptime,
							&args[index] );
					continue;
				case GET_MEMORY:
					pthread_create( &job_thread[index],
							NULL,
							get_memory,
							&args[index] );
					continue;
				default:
					buf.chunk.id = -1;
					sprintf( (char*)buf.chunk.data,
						 "Unsupported job type with ID: %d",
						 buf.meta.type );
					SSL_write(ssl, &buf, sizeof(struct msg_t));
					continue;
				}
			} else {
				index = find_job(buf.meta.id);
				if (index == FAIL) {
					sprintf( (char*)buf.chunk.data,
						 "Data was sent for an invalid job ID" );
					SSL_write(ssl, &buf, sizeof(struct msg_t));
				} else
					memcpy(&args[index].buf, &buf, sizeof(struct msg_t));
			}
		}
		
		SSL_shutdown(ssl);
		SSL_free(ssl);		/* release connection state */
	} while (bytes);
	close(server);			/* close socket */
	SSL_CTX_free(ctx);		/* release context */
}