Initial commit. It's far from finished.
This commit is contained in:
167
thread_pool.c
Normal file
167
thread_pool.c
Normal file
@@ -0,0 +1,167 @@
|
||||
#include "thread_pool.h"
|
||||
#include "log_trace.h"
|
||||
#include "protocol.h"
|
||||
#include <pthread.h>
|
||||
#include <unistd.h>
|
||||
|
||||
#define MAXJOBS 10
|
||||
|
||||
struct agent_args {
|
||||
int busy;
|
||||
SSL *ssl;
|
||||
int sd;
|
||||
char ip[16]; /* IPv4 */
|
||||
};
|
||||
|
||||
static void show_certs(SSL *ssl);
|
||||
static void* servlet(void *args);
|
||||
static void send_reject_msg(SSL *ssl);
|
||||
|
||||
static void show_certs(SSL *ssl)
|
||||
{
|
||||
X509 *cert;
|
||||
char *line;
|
||||
|
||||
cert = SSL_get_peer_certificate(ssl); /* Get certificates (if available) */
|
||||
if (SSL_get_verify_result(ssl)==X509_V_OK)
|
||||
log_trace(VERBOSE, "get_verify_result == ok");
|
||||
if (cert != NULL) {
|
||||
log_trace(VERBOSE, "Server certificates:");
|
||||
line = X509_NAME_oneline(X509_get_subject_name(cert), 0, 0);
|
||||
log_trace(VERBOSE, "Subject: %s", line);
|
||||
free(line);
|
||||
line = X509_NAME_oneline(X509_get_issuer_name(cert), 0, 0);
|
||||
log_trace(VERBOSE, "Issuer: %s", line);
|
||||
free(line);
|
||||
X509_free(cert);
|
||||
} else
|
||||
log_trace(VERBOSE, "No certificates from peer");
|
||||
}
|
||||
|
||||
static void* servlet(void *args) /* Serve the connection -- threadable */
|
||||
{
|
||||
struct msg buf;
|
||||
char reply[2048];
|
||||
int bytes, ret;
|
||||
//unsigned short job[MAXJOBS] = { 0 };
|
||||
struct agent_args *agent = (struct agent_args*)args;
|
||||
|
||||
SSL_load_error_strings();
|
||||
ret = SSL_accept(agent->ssl);
|
||||
/* We check for unclean (ret < 0) and clean (ret == 0) failures */
|
||||
if (ret <= 0) {
|
||||
char ret_str[1024];
|
||||
ERR_error_string_n(SSL_get_error(agent->ssl, ret), ret_str, sizeof(ret_str));
|
||||
log_trace(INFO, "SSL_accept() failed. Reason below:\n%s", ret_str);
|
||||
} else {
|
||||
show_certs(agent->ssl);
|
||||
do {
|
||||
buf.meta.type = GET_MEMORY;
|
||||
sleep(1);
|
||||
SSL_write(agent->ssl, &buf, sizeof(buf));
|
||||
bytes = SSL_read(agent->ssl, &buf, sizeof(buf));
|
||||
if (bytes > 0) {
|
||||
if (bytes != sizeof(struct msg)) {
|
||||
log_trace( WARNING,
|
||||
"Agent [%s] sent non-standard data!",
|
||||
agent->ip );
|
||||
continue;
|
||||
}
|
||||
//buf.chunk.data[bytes] = 0;
|
||||
|
||||
log_trace(VERBOSE, "Client msg: \"%s\"", buf.chunk.data);
|
||||
SSL_write(agent->ssl, buf.chunk.data, buf.meta.len); /* send reply */
|
||||
continue;
|
||||
}
|
||||
if (SSL_get_shutdown(agent->ssl) == SSL_RECEIVED_SHUTDOWN)
|
||||
log_trace(VERBOSE, "SSL_RECEIVED_SHUTDOWN from agent [%s]", agent->ip);
|
||||
else {
|
||||
char ssl_errstr[2048];
|
||||
long ssl_errnum = ERR_get_error();
|
||||
ERR_error_string_n(ssl_errnum, ssl_errstr, sizeof(ssl_errstr));
|
||||
ERR_print_errors_fp(stderr);
|
||||
log_trace( VERBOSE,
|
||||
"Client didn't send data! SSL error below:\n%s",
|
||||
ssl_errstr);
|
||||
sprintf(reply, "%s", "Where's the data, m8?");
|
||||
SSL_write(agent->ssl, reply, strlen(reply));
|
||||
}
|
||||
log_trace(INFO, "Agent [%s] disconnected.", agent->ip);
|
||||
} while (bytes);
|
||||
}
|
||||
SSL_free(agent->ssl); /* release SSL state */
|
||||
close(agent->sd); /* close connection */
|
||||
agent->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 ssl_pt_mutex(int srv, SSL_CTX *ctx, int poolsize)
|
||||
{
|
||||
pthread_mutex_t mutex;
|
||||
pthread_attr_t attr;
|
||||
pthread_t *agent_thread = (pthread_t*)malloc(poolsize * sizeof(pthread_t));
|
||||
struct agent_args *agent_struct =
|
||||
(struct agent_args*)malloc(poolsize * sizeof(struct agent_args));
|
||||
int i;
|
||||
|
||||
memset(agent_thread, 0, sizeof(pthread_t) * poolsize);
|
||||
memset(agent_struct, 0, sizeof(struct agent_args) * poolsize);
|
||||
|
||||
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(srv, (struct sockaddr*)&addr, &len);
|
||||
log_trace( VERBOSE,
|
||||
"Connection: %s:%d",
|
||||
inet_ntop(AF_INET, &addr.sin_addr, address, sizeof(address)),
|
||||
ntohs(addr.sin_port)
|
||||
);
|
||||
|
||||
for (i = 0; i < poolsize; i++) {
|
||||
if (!agent_struct[i].busy) {
|
||||
agent_struct[i].busy = 1;
|
||||
agent_struct[i].ssl = SSL_new(ctx);
|
||||
agent_struct[i].sd = agent;
|
||||
memcpy( agent_struct[i].ip,
|
||||
inet_ntop(AF_INET, &addr.sin_addr, address, sizeof(address)),
|
||||
sizeof(agent_struct[i].ip) );
|
||||
SSL_set_fd(agent_struct[i].ssl, agent_struct[i].sd);
|
||||
pthread_create( &agent_thread[i],
|
||||
&attr,
|
||||
servlet,
|
||||
&agent_struct[i] );
|
||||
break;
|
||||
}
|
||||
}
|
||||
if (i == poolsize) {
|
||||
log_trace( WARNING,
|
||||
"Agent [%s] dropped. Poolsize limit reached.",
|
||||
inet_ntop(AF_INET, &addr.sin_addr, address, sizeof(address))
|
||||
);
|
||||
ssl = SSL_new(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);
|
||||
}
|
||||
Reference in New Issue
Block a user