smirky/rmps
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Start working on client listener

Browse Source
This commit is contained in:
Bogomil Vasilev
2016-08-29 17:21:20 +03:00
parent fa16b4fb8c
commit 97d7ec8b13
11 changed files with 322 additions and 77 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
Makefile
View File

@@ -15,7 +15,8 @@ SOURCES = main.c \
rmps.c \
enum_functions.c \
log.c \
thread_pool.c
agent_pool.c \
client_pool.c
OBJECTS = $(SOURCES:.c=.o)
EXECUTABLE = rmpsd

28
thread_pool.c → agent_pool.c
View File

@@ -1,4 +1,4 @@
#include "thread_pool.h"
#include "agent_pool.h"
#include "log.h"
#include "protocol.h"
#include <pthread.h>
@@ -94,17 +94,18 @@ static void send_reject_msg(SSL *ssl)
SSL_write(ssl, reply, strlen(reply));
}
void ssl_pt_mutex(int srv, SSL_CTX *ctx, int poolsize)
void* agent_pool(void *args)
{
struct pool_data *pool = args;
pthread_mutex_t mutex;
pthread_attr_t attr;
pthread_t *agent_thread = (pthread_t*)malloc(poolsize * sizeof(pthread_t));
pthread_t *agent_thread = (pthread_t*)malloc(pool->size * sizeof(pthread_t));
struct agent_args *agent_struct =
(struct agent_args*)malloc(poolsize * sizeof(struct agent_args));
(struct agent_args*)malloc(pool->size * sizeof(struct agent_args));
int i;
memset(agent_thread, 0, sizeof(pthread_t) * poolsize);
memset(agent_struct, 0, sizeof(struct agent_args) * poolsize);
memset(agent_thread, 0, sizeof(pthread_t) * pool->size);
memset(agent_struct, 0, sizeof(struct agent_args) * pool->size);
pthread_mutex_init(&mutex, NULL);
pthread_attr_init(&attr);
@@ -115,17 +116,16 @@ void ssl_pt_mutex(int srv, SSL_CTX *ctx, int poolsize)
char address[INET6_ADDRSTRLEN];
socklen_t len = sizeof(addr);
SSL *ssl;
int agent = accept(srv, (struct sockaddr*)&addr, &len);
log( INFO,
int agent = accept(pool->srv, (struct sockaddr*)&addr, &len);
log(INFO,
"Connection: %s:%d",
inet_ntop(AF_INET, &addr.sin_addr, address, sizeof(address)),
ntohs(addr.sin_port)
);
ntohs(addr.sin_port));
for (i = 0; i < poolsize; i++) {
for (i = 0; i < pool->size; i++) {
if (!agent_struct[i].busy) {
agent_struct[i].busy = 1;
agent_struct[i].ssl = SSL_new(ctx);
agent_struct[i].ssl = SSL_new(pool->ctx);
agent_struct[i].sd = agent;
memcpy( agent_struct[i].ip,
inet_ntop(AF_INET, &addr.sin_addr, address, sizeof(address)),
@@ -138,12 +138,12 @@ void ssl_pt_mutex(int srv, SSL_CTX *ctx, int poolsize)
break;
}
}
if (i == poolsize) {
if (i == pool->size) {
log( WARNING,
"Agent [%s] dropped. Poolsize limit reached.",
inet_ntop(AF_INET, &addr.sin_addr, address, sizeof(address))
);
ssl = SSL_new(ctx);
ssl = SSL_new(pool->ctx);
SSL_set_fd(ssl, agent);
if (SSL_accept(ssl) == FAIL) {
SSL_free(ssl);

10
thread_pool.h → agent_pool.h
View File

@@ -1,5 +1,5 @@
#ifndef THREAD_POOL_H
#define THREAD_POOL_H
#ifndef AGENT_POOL_H
#define AGENT_POOL_H
/* included for openssl and sockets */
#include <sys/socket.h>
@@ -7,9 +7,11 @@
#include <resolv.h>
#include <openssl/ssl.h>
#include <openssl/err.h>
#include "rmps.h"
#define FAIL -1
void ssl_pt_mutex(int srv, SSL_CTX *ctx, int poolsize);
//void agent_pool(int srv, SSL_CTX *ctx, int poolsize);
void* agent_pool(void *args);
#endif /* THREAD_POOL_H */
#endif /* AGENT_POOL_H */

137
client_pool.c Normal file
View File

@@ -0,0 +1,137 @@
#include "client_pool.h"
#include "log.h"
#include "protocol.h"
#include <pthread.h>
#include <unistd.h>
#define MAXJOBS 10
struct client_args {
int busy;
SSL *ssl;
int sd;
char ip[16]; /* IPv4 */
};
static void* servlet(void *args);
static void send_reject_msg(SSL *ssl);
static void* servlet(void *args) /* Serve the connection -- threadable */
{
struct msg_t buf;
int bytes, ret;
//unsigned short job[MAXJOBS] = { 0 };
struct client_args *client = (struct client_args*)args;
SSL_load_error_strings();
ret = SSL_accept(client->ssl);
/* We check for unclean (ret < 0) and clean (ret == 0) failures */
if (ret <= 0) {
log(WARNING, "SSL_accept() failed. Reason below:");
log_ssl();
} else {
do {
//buf.meta.type = GET_MEMORY;
//sleep(1);
//SSL_write(client->ssl, &buf, sizeof(buf));
bytes = SSL_read(client->ssl, &buf, sizeof(buf));
if (bytes > 0) {
if (bytes != sizeof(struct msg_t)) {
log(WARNING,
"Client [%s] sent non-standard data!",
client->ip );
continue;
}
log(VERBOSE, "Client msg: \"%s\"", buf.chunk.data);
/* TODO: Insert msg handler here */
continue;
}
if (SSL_get_shutdown(client->ssl) == SSL_RECEIVED_SHUTDOWN)
log(VERBOSE, "SSL_RECEIVED_SHUTDOWN from client [%s]", client->ip);
else {
log(VERBOSE, "Client didn't send data! SSL error below:");
//log_ssl(); /* We actually don't have anything to log from SSL */
sprintf((char*)buf.chunk.data, "%s", "Where's the data, m8?");
SSL_write(client->ssl, &buf, sizeof(struct msg_t));
}
log(INFO, "Client [%s] disconnected.", client->ip);
} while (bytes);
}
SSL_free(client->ssl); /* release SSL state */
close(client->sd); /* close connection */
client->busy = 0;
return 0;
}
static void send_reject_msg(SSL *ssl)
{
char *reply = "FAILURE - The connection queue is full!\n";
SSL_write(ssl, reply, strlen(reply));
}
void* client_pool(void *args)
{
struct pool_data *pool = args;
pthread_mutex_t mutex;
pthread_attr_t attr;
pthread_t *client_thread = (pthread_t*)malloc(pool->size * sizeof(pthread_t));
struct client_args *client_struct =
(struct client_args*)malloc(pool->size * sizeof(struct client_args));
int i;
memset(client_thread, 0, sizeof(pthread_t) * pool->size);
memset(client_struct, 0, sizeof(struct client_args) * pool->size);
pthread_mutex_init(&mutex, NULL);
pthread_attr_init(&attr);
pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_JOINABLE);
while (1) {
struct sockaddr_in addr;
char address[INET6_ADDRSTRLEN];
socklen_t len = sizeof(addr);
SSL *ssl;
int agent = accept(pool->srv, (struct sockaddr*)&addr, &len);
log(INFO,
"Connection: %s:%d",
inet_ntop(AF_INET, &addr.sin_addr, address, sizeof(address)),
ntohs(addr.sin_port));
for (i = 0; i < pool->size; i++) {
if (!client_struct[i].busy) {
client_struct[i].busy = 1;
client_struct[i].ssl = SSL_new(pool->ctx);
client_struct[i].sd = agent;
memcpy( client_struct[i].ip,
inet_ntop(AF_INET, &addr.sin_addr, address, sizeof(address)),
sizeof(client_struct[i].ip) );
SSL_set_fd(client_struct[i].ssl, client_struct[i].sd);
pthread_create( &client_thread[i],
&attr,
servlet,
&client_struct[i] );
break;
}
}
if (i == pool->size) {
log( WARNING,
"Agent [%s] dropped. Poolsize limit reached.",
inet_ntop(AF_INET, &addr.sin_addr, address, sizeof(address))
);
ssl = SSL_new(pool->ctx);
SSL_set_fd(ssl, agent);
if (SSL_accept(ssl) == FAIL) {
SSL_free(ssl);
close(agent);
continue;
}
send_reject_msg(ssl);
SSL_free(ssl);
close(agent);
}
}
pthread_attr_destroy(&attr);
pthread_mutex_destroy(&mutex);
}

17
client_pool.h Normal file
View File

@@ -0,0 +1,17 @@
#ifndef CLIENT_POOL_H
#define CLIENT_POOL_H
/* included for openssl and sockets */
#include <sys/socket.h>
#include <arpa/inet.h>
#include <resolv.h>
#include <openssl/ssl.h>
#include <openssl/err.h>
#include "rmps.h"
#define FAIL -1
//void client_pool(int srv, SSL_CTX *ctx, int poolsize);
void* client_pool(void *args);
#endif /* CLIENT_POOL_H */

108
confparser.c
View File

@@ -19,17 +19,22 @@ struct conf_table conf = {
"" /* db.port */
},
{
"127.0.0.1", /* rmps.bind_on_ip */
"7000", /* rmps.bind_on_port */
"127.0.0.1", /* rmps.agent_ip */
"7000", /* rmps.agent_port */
"127.0.0.1", /* rmps.client_ip */
"7001", /* rmps.client_port */
"/var/log/rmps/rmpsd.log",
"/var/log/rmps/rmpsd.err",
'2', /* rmps.loglevel */
"/run/rmps/rmpsd.pid",
"/etc/rmps/cert.pem",
"/etc/rmps/key.pem",
"/etc/rmps/agent.crt",
"/etc/rmps/agent.key",
"/etc/rmps/ca.crt",
"", /* rmps.cipherlist */
2 /* rmps.threadpoolsize */
2, /* rmps.agent_poolsize */
"/etc/rmps/client.crt",
"/etc/rmps/client.key",
2 /* rmps.client_poolsize */
},
{
0 /* nfs -> TODO */
@@ -41,31 +46,41 @@ void confexport(void)
printf( "db.type=%s\n"
"db.hostname=%s\n"
"db.port=%s\n"
"rmps.bind_on_ip=%s\n"
"rmps.bind_on_port=%s\n"
"rmps.agent_ip=%s\n"
"rmps.agent_port=%s\n"
"rmps.client_ip=%s\n"
"rmps.client_port=%s\n"
"rmps.logfile=%s\n"
"rmps.errlog=%s\n"
"rmps.loglevel=%c\n"
"rmps.loglevel=%d\n"
"rmps.pidfile=%s\n"
"rmps.certfile=%s\n"
"rmps.keyfile=%s\n"
"rmps.agent_tls_crt=%s\n"
"rmps.agent_tls_key=%s\n"
"rmps.cafile=%s\n"
"rmps.cipherlist=%s\n"
"rmps.threadpoolsize=%d\n",
"rmps.agent_poolsize=%d\n"
"rmps.client_tls_crt=%s\n"
"rmps.client_tls_key=%s\n"
"rmps.client_poolsize=%d\n",
conf.db.type,
conf.db.hostname,
conf.db.port,
conf.rmps.bind_on_ip,
conf.rmps.bind_on_port,
conf.rmps.agent_ip,
conf.rmps.agent_port,
conf.rmps.client_ip,
conf.rmps.client_port,
conf.rmps.logfile,
conf.rmps.errlog,
conf.rmps.loglevel,
conf.rmps.pidfile,
conf.rmps.certfile,
conf.rmps.keyfile,
conf.rmps.agent_tls_crt,
conf.rmps.agent_tls_key,
conf.rmps.cafile,
conf.rmps.cipherlist,
conf.rmps.threadpoolsize
conf.rmps.agent_poolsize,
conf.rmps.client_tls_crt,
conf.rmps.client_tls_key,
conf.rmps.client_poolsize
);
}
@@ -217,7 +232,7 @@ static int test_conf_syntax(void)
continue;
}
/* Here we check every single entry manually */
/* Here we check every single conf entry manually */
if (!strcmp(buf, "db.type")) {
if (!strcmp(tmp + 1, "mysql")) {
/* || !strcmp(tmp[1], "postgresql") */
@@ -244,15 +259,28 @@ static int test_conf_syntax(void)
}
ok = 0;
failed = 1;
} else if (!strcmp(buf, "rmps.bind_on_ip")) {
} else if (!strcmp(buf, "rmps.agent_ip")) {
/* TODO */
} else if (!strcmp(buf, "rmps.bind_on_port")) {
} else if (!strcmp(buf, "rmps.agent_port")) {
if ((i = strlen(tmp + 1)) < 6) {
if ((signed int)strspn(tmp + 1, "1234567890") == i) {
i = atoi(tmp + 1);
if (i > 0 && i < 65536) {
strcpy(conf.rmps.bind_on_port, tmp + 1);
strcpy(conf.rmps.agent_port, tmp + 1);
continue;
}
}
}
ok = 0;
failed = 1;
} else if (!strcmp(buf, "rmps.client_ip")) {
/* TODO */
} else if (!strcmp(buf, "rmps.client_port")) {
if ((i = strlen(tmp + 1)) < 6) {
if ((signed int)strspn(tmp + 1, "1234567890") == i) {
i = atoi(tmp + 1);
if (i > 0 && i < 65536) {
strcpy(conf.rmps.client_port, tmp + 1);
continue;
}
}
@@ -276,7 +304,7 @@ static int test_conf_syntax(void)
conf.rmps.loglevel = tmp[1] - '0';
else
failed = 1;
} else if (!strcmp(buf, "rmps.certfile")) {
} else if (!strcmp(buf, "rmps.agent_tls_crt")) {
if (access(tmp + 1, F_OK) == -1) {
log(ERROR, "%s is missing", tmp + 1);
failed = 1;
@@ -285,18 +313,22 @@ static int test_conf_syntax(void)
log(ERROR, "%s is not readable", tmp + 1);
failed = 1;
} else
strncpy(conf.rmps.certfile, tmp + 1, sizeof(conf.rmps.certfile));
strncpy(conf.rmps.agent_tls_crt,
tmp + 1,
sizeof(conf.rmps.agent_tls_crt));
}
else if (!strcmp(buf, "rmps.keyfile")) {
else if (!strcmp(buf, "rmps.agent_tls_key")) {
if (access(tmp + 1, F_OK) == -1) {
log(ERROR, "%s is missing", conf.rmps.keyfile);
log(ERROR, "%s is missing", tmp + 1);
failed = 1;
}
else if (access(tmp + 1, R_OK) == -1) {
log(ERROR, "%s is not readable", tmp + 1);
failed = 1;
} else
strncpy(conf.rmps.keyfile, tmp + 1, sizeof(conf.rmps.keyfile));
strncpy(conf.rmps.agent_tls_key,
tmp + 1,
sizeof(conf.rmps.agent_tls_key));
} else if (!strcmp(buf, "rmps.cipherlist")) {
strncpy(conf.rmps.cipherlist, tmp + 1, sizeof(conf.rmps.cipherlist));
} else if (!strcmp(buf, "rmps.cafile")) {
@@ -309,6 +341,30 @@ static int test_conf_syntax(void)
failed = 1;
} else
strncpy(conf.rmps.cafile, tmp + 1, sizeof(conf.rmps.cafile));
} else if (!strcmp(buf, "rmps.client_tls_crt")) {
if (access(tmp + 1, F_OK) == -1) {
log(ERROR, "%s is missing", tmp + 1);
failed = 1;
}
else if (access(tmp + 1, R_OK) == -1) {
log(ERROR, "%s is not readable", tmp + 1);
failed = 1;
} else
strncpy(conf.rmps.client_tls_crt,
tmp + 1,
sizeof(conf.rmps.client_tls_crt));
} else if (!strcmp(buf, "rmps.client_tls_key")) {
if (access(tmp + 1, F_OK) == -1) {
log(ERROR, "%s is missing", tmp + 1);
failed = 1;
}
else if (access(tmp + 1, R_OK) == -1) {
log(ERROR, "%s is not readable", tmp + 1);
failed = 1;
} else
strncpy(conf.rmps.client_tls_key,
tmp + 1,
sizeof(conf.rmps.client_tls_key));
} else
log(ERROR, "Unknown config entry on line %d: %s", j, buf);
if (!ok) {

15
confparser.h
View File

@@ -14,17 +14,22 @@ struct conf_db {
};
struct conf_rmps {
char bind_on_ip[13];
char bind_on_port[6];
char agent_ip[13];
char agent_port[6];
char client_ip[13];
char client_port[6];
char logfile[MAXPATHSIZE];
char errlog[MAXPATHSIZE];
LOG_LEVEL loglevel;
char pidfile[MAXPATHSIZE];
char certfile[MAXPATHSIZE];
char keyfile[MAXPATHSIZE];
char agent_tls_crt[MAXPATHSIZE];
char agent_tls_key[MAXPATHSIZE];
char cafile[MAXPATHSIZE];
char cipherlist[1024];
int threadpoolsize;
int agent_poolsize;
char client_tls_crt[MAXPATHSIZE];
char client_tls_key[MAXPATHSIZE];
int client_poolsize;
};
struct conf_nfs {

2
log.c
View File

@@ -51,7 +51,7 @@ void log(LOG_LEVEL lvl, char *fmt, ... )
time_t t = time(NULL);
struct tm tm;
localtime_r(&t, &tm);
if (conf.rmps.loglevel == ERROR || conf.rmps.loglevel == WARNING)
if (lvl == ERROR || lvl == WARNING)
fp = fderr;
else
fp = fdout;

46
rmps.c
View File

@@ -1,7 +1,9 @@
#include "log.h"
#include "confparser.h"
#include "thread_pool.h"
#include "agent_pool.h"
#include "client_pool.h"
#include "rmps.h"
#include <pthread.h>
#include <unistd.h>
#include <sys/stat.h>
#include <fcntl.h>
@@ -197,8 +199,8 @@ static SSL_CTX* init_server_ctx(const char *cipherlist)
SSL_CTX *ctx;
char ciphers[1024];
OpenSSL_add_all_algorithms(); /* load & register all cryptos, etc. */
OpenSSL_add_all_ciphers(); /* load & register all cryptos, etc. */
// 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();
@@ -257,11 +259,12 @@ void load_certificates(SSL_CTX* ctx, const char *certfile,
}
int launch_rmps(struct conf_table *conf, int fork_flag)
void launch_rmps(struct conf_table *conf, int fork_flag)
{
int server;
log(INFO, "Starting up RMPS...");
pthread_t pool[2];
struct pool_data pool_args[2];
log(INFO, "Starting up RMPS...");
/* Set signal handling */
set_env();
/* Deamonize */
@@ -270,20 +273,31 @@ int launch_rmps(struct conf_table *conf, int fork_flag)
/* Spawn & lock pidfile */
spawn_pidfile(conf->rmps.pidfile);
SSL_CTX *ctx;
ctx = init_server_ctx(conf->rmps.cipherlist);
/* 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
*/
log(VERBOSE, "Loading crypto certs and keys.");
load_certificates(ctx, conf->rmps.certfile, conf->rmps.keyfile, conf->rmps.cafile);
pool_args[0].ctx = init_server_ctx(conf->rmps.cipherlist);
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]);
log(VERBOSE, "Starting listener on port: %d", atoi(conf->rmps.bind_on_port));
server = open_listener(atoi(conf->rmps.bind_on_port));
pool_args[1].ctx = init_server_ctx(conf->rmps.cipherlist);
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]);
log(VERBOSE, "Creating mutex for thread pool.");
ssl_pt_mutex(server, ctx, conf->rmps.threadpoolsize);
return 0;
pthread_join(pool[0], NULL);
pthread_join(pool[1], NULL);
}

18
rmps.conf
View File

@@ -12,19 +12,23 @@ db.hostname=127.0.0.1
db.port=3306
# RMPS Core settings
rmps.bind_on_ip=127.0.0.1
rmps.bind_on_port=7000
rmps.agent_ip=127.0.0.1
rmps.agent_port=7000
rmps.client_ip=127.0.0.1
rmps.client_port=7001
# Loglevel modes: 1:ERROR,2:WARNING,3:INFO,4:VERBOSE
# The default is 2.
rmps.loglevel=4
rmps.logfile=/home/smirky/stuff/projects/rmps/tmp/rmpsd.log
rmps.errlog=/home/smirky/stuff/projects/rmps/tmp/rmpsd.err
rmps.pidfile=/home/smirky/stuff/projects/rmps/tmp/rmpsd.pig
rmps.logfile=/home/smirky/projects/rmps/tmp/rmpsd.log
rmps.errlog=/home/smirky/projects/rmps/tmp/rmpsd.err
rmps.pidfile=/home/smirky/projects/rmps/tmp/rmpsd.pig
#rmps.pidfile=/home/smirky/stuff/projects/rmps/rmpsd/rmpsd.pid
# Cryptography settings
rmps.certfile=/etc/rmps/certs/server/server.crt
rmps.keyfile=/etc/rmps/certs/server/server.key
rmps.agent_tls_crt=/etc/rmps/certs/server/server.crt
rmps.agent_tls_key=/etc/rmps/certs/server/server.key
rmps.client_tls_crt=/etc/rmps/certs/server/server.crt
rmps.client_tls_key=/etc/rmps/certs/server/server.key
rmps.cafile=/etc/rmps/certs/ca.pem
rmps.cipherlist=ECDHE-RSA-AES128-SHA256:AES256-SHA:RSA

11
rmps.h
View File

@@ -1,6 +1,15 @@
#ifndef RMPS_H
#define RMPS_H
extern int launch_rmps(struct conf_table *conf, int fork_flag);
#include <openssl/ssl.h>
#include "confparser.h"
struct pool_data {
int srv;
SSL_CTX *ctx;
int size;
};
extern void launch_rmps(struct conf_table *conf, int fork_flag);
#endif /* RMPS_H */