diff options
author | Martin Willi <martin@strongswan.org> | 2006-05-10 08:02:49 +0000 |
---|---|---|
committer | Martin Willi <martin@strongswan.org> | 2006-05-10 08:02:49 +0000 |
commit | b8577029d1d37b798907f0418ddb1445e13c3c44 (patch) | |
tree | 6bc5eaa1db913fdf6b6541a177607a7aa79f4dfc /src/charon/threads | |
parent | 95806de938a287ca71b28fa07016c9785130c1da (diff) | |
download | strongswan-b8577029d1d37b798907f0418ddb1445e13c3c44.tar.bz2 strongswan-b8577029d1d37b798907f0418ddb1445e13c3c44.tar.xz |
Diffstat (limited to 'src/charon/threads')
-rw-r--r-- | src/charon/threads/Makefile.threads | 39 | ||||
-rw-r--r-- | src/charon/threads/kernel_interface.c | 729 | ||||
-rw-r--r-- | src/charon/threads/kernel_interface.h | 185 | ||||
-rw-r--r-- | src/charon/threads/receiver.c | 128 | ||||
-rw-r--r-- | src/charon/threads/receiver.h | 67 | ||||
-rw-r--r-- | src/charon/threads/scheduler.c | 124 | ||||
-rw-r--r-- | src/charon/threads/scheduler.h | 67 | ||||
-rw-r--r-- | src/charon/threads/sender.c | 126 | ||||
-rw-r--r-- | src/charon/threads/sender.h | 63 | ||||
-rwxr-xr-x | src/charon/threads/stroke_interface.c | 646 | ||||
-rw-r--r-- | src/charon/threads/stroke_interface.h | 76 | ||||
-rw-r--r-- | src/charon/threads/thread_pool.c | 623 | ||||
-rw-r--r-- | src/charon/threads/thread_pool.h | 78 |
13 files changed, 2951 insertions, 0 deletions
diff --git a/src/charon/threads/Makefile.threads b/src/charon/threads/Makefile.threads new file mode 100644 index 000000000..949c1ad24 --- /dev/null +++ b/src/charon/threads/Makefile.threads @@ -0,0 +1,39 @@ +# Copyright (C) 2005 Jan Hutter, Martin Willi +# Hochschule fuer Technik Rapperswil +# +# This program is free software; you can redistribute it and/or modify it +# under the terms of the GNU General Public License as published by the +# Free Software Foundation; either version 2 of the License, or (at your +# option) any later version. See <http://www.fsf.org/copyleft/gpl.txt>. +# +# 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 General Public License +# for more details. +# + +THREADS_DIR= $(CHARON_DIR)threads/ + +CHARON_OBJS+= $(BUILD_DIR)receiver.o +$(BUILD_DIR)receiver.o : $(THREADS_DIR)receiver.c $(THREADS_DIR)receiver.h + $(CC) $(CFLAGS) -c -o $@ $< + +CHARON_OBJS+= $(BUILD_DIR)scheduler.o +$(BUILD_DIR)scheduler.o : $(THREADS_DIR)scheduler.c $(THREADS_DIR)scheduler.h + $(CC) $(CFLAGS) -c -o $@ $< + +CHARON_OBJS+= $(BUILD_DIR)sender.o +$(BUILD_DIR)sender.o : $(THREADS_DIR)sender.c $(THREADS_DIR)sender.h + $(CC) $(CFLAGS) -c -o $@ $< + +CHARON_OBJS+= $(BUILD_DIR)thread_pool.o +$(BUILD_DIR)thread_pool.o : $(THREADS_DIR)thread_pool.c $(THREADS_DIR)thread_pool.h + $(CC) $(CFLAGS) -c -o $@ $< + +CHARON_OBJS+= $(BUILD_DIR)kernel_interface.o +$(BUILD_DIR)kernel_interface.o :$(THREADS_DIR)kernel_interface.c $(THREADS_DIR)kernel_interface.h + $(CC) $(CFLAGS) -c -o $@ $< + +CHARON_OBJS+= $(BUILD_DIR)stroke_interface.o +$(BUILD_DIR)stroke_interface.o :$(THREADS_DIR)stroke_interface.c $(THREADS_DIR)stroke_interface.h + $(CC) $(CFLAGS) -c -o $@ $< diff --git a/src/charon/threads/kernel_interface.c b/src/charon/threads/kernel_interface.c new file mode 100644 index 000000000..679cf69ee --- /dev/null +++ b/src/charon/threads/kernel_interface.c @@ -0,0 +1,729 @@ +/** + * @file kernel_interface.c + * + * @brief Implementation of kernel_interface_t. + * + */ + +/* + * Copyright (C) 2005 Jan Hutter, Martin Willi + * Hochschule fuer Technik Rapperswil + * Copyright (C) 2003 Herbert Xu. + * + * Contains modified parts from pluto. + * + * This program is free software; you can redistribute it and/or modify it + * under the terms of the GNU General Public License as published by the + * Free Software Foundation; either version 2 of the License, or (at your + * option) any later version. See <http://www.fsf.org/copyleft/gpl.txt>. + * + * 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 General Public License + * for more details. + */ + +#include <sys/types.h> +#include <sys/socket.h> +#include <linux/netlink.h> +#include <pthread.h> +#include <unistd.h> +#include <fcntl.h> +#include <errno.h> +#include <string.h> + +#include "kernel_interface.h" + +#include <daemon.h> +#include <utils/linked_list.h> + + +#define KERNEL_ESP 50 +#define KERNEL_AH 51 + +#define SPD_PRIORITY 1024 + +#define XFRM_DATA_LENGTH 512 + + +typedef struct xfrm_data_t xfrm_data_t; + +/** + * Lenght/Type/data struct for userdata in xfrm + * We dont use the "I-don't-know-where-they-come-from"-structs + * used in the kernel. + */ +struct xfrm_data_t { + /** + * length of the data + */ + u_int16_t length; + + /** + * type of data + */ + u_int16_t type; + + /** + * and the data itself, for different purposes + */ + union { + /** algorithm */ + struct xfrm_algo algo; + /** policy tmpl */ + struct xfrm_user_tmpl tmpl[2]; + }; +}; + + +typedef struct netlink_message_t netlink_message_t; + +/** + * Representation of ANY netlink message used + */ +struct netlink_message_t { + + /** + * header of the netlink message + */ + struct nlmsghdr hdr; + + union { + /** error message */ + struct nlmsgerr e; + /** message for spi allocation */ + struct xfrm_userspi_info spi; + /** message for SA manipulation */ + struct xfrm_usersa_id sa_id; + /** message for SA installation */ + struct xfrm_usersa_info sa; + /** message for policy manipulation */ + struct xfrm_userpolicy_id policy_id; + /** message for policy installation */ + struct xfrm_userpolicy_info policy; + }; + u_int8_t data[XFRM_DATA_LENGTH]; +}; + + +typedef struct private_kernel_interface_t private_kernel_interface_t; + + /** + * @brief Private Variables and Functions of kernel_interface class. + * + */ +struct private_kernel_interface_t { + /** + * Public part of the kernel_interface_t object. + */ + kernel_interface_t public; + + /** + * Netlink communication socket. + */ + int socket; + + /** + * Process id of kernel thread + */ + pid_t pid; + + /** + * Sequence number for messages. + */ + u_int32_t seq; + + /** + * List of responded messages. + */ + linked_list_t *responses; + + /** + * Thread which receives messages. + */ + pthread_t thread; + + /** + * Mutex locks access to replies list. + */ + pthread_mutex_t mutex; + + /** + * Condvar allows signaling of threads waiting for a reply. + */ + pthread_cond_t condvar; + + /** + * Logger for XFRM stuff + */ + logger_t *logger; + + /** + * Function for the thread, receives messages. + */ + void (*receive_messages) (private_kernel_interface_t *this); + + /** + * Sends a netlink_message_t down to the kernel and wait for reply. + */ + status_t (*send_message) (private_kernel_interface_t *this, netlink_message_t *request, netlink_message_t **response); +}; + +/** + * In the kernel, algorithms are identified as strings, we use our + * mapping functions... + * Algorithms for encryption. + * TODO: Add missing algorithm strings + */ +mapping_t kernel_encryption_algs_m[] = { + {ENCR_DES_IV64, ""}, + {ENCR_DES, "des"}, + {ENCR_3DES, "des3_ede"}, + {ENCR_RC5, ""}, + {ENCR_IDEA, "idea"}, + {ENCR_CAST, "cast128"}, + {ENCR_BLOWFISH, "blowfish"}, + {ENCR_3IDEA, ""}, + {ENCR_DES_IV32, ""}, + {ENCR_NULL, ""}, + {ENCR_AES_CBC, "aes"}, + {ENCR_AES_CTR, ""}, + {MAPPING_END, NULL} +}; +/** + * In the kernel, algorithms are identified as strings, we use our + * mapping functions... + * Algorithms for integrity protection. + * TODO: Add missing algorithm strings + */ +mapping_t kernel_integrity_algs_m[] = { + {AUTH_HMAC_MD5_96, "md5"}, + {AUTH_HMAC_SHA1_96, "sha1"}, + {AUTH_DES_MAC, ""}, + {AUTH_KPDK_MD5, ""}, + {AUTH_AES_XCBC_96, ""}, + {MAPPING_END, NULL} +}; + + +/** + * Implementation of kernel_interface_t.get_spi. + */ +static status_t get_spi(private_kernel_interface_t *this, + host_t *src, host_t *dest, + protocol_id_t protocol, u_int32_t reqid, + u_int32_t *spi) +{ + netlink_message_t request, *response; + status_t status = SUCCESS; + + + this->logger->log(this->logger, CONTROL|LEVEL2, "getting spi"); + + memset(&request, 0, sizeof(request)); + request.hdr.nlmsg_len = NLMSG_ALIGN(NLMSG_LENGTH(sizeof(request.spi))); + request.hdr.nlmsg_flags = NLM_F_REQUEST; + request.hdr.nlmsg_type = XFRM_MSG_ALLOCSPI; + request.spi.info.saddr = src->get_xfrm_addr(src); + request.spi.info.id.daddr = dest->get_xfrm_addr(dest); + request.spi.info.mode = TRUE; /* tunnel mode */ + request.spi.info.reqid = reqid; + request.spi.info.id.proto = (protocol == PROTO_ESP) ? KERNEL_ESP : KERNEL_AH; + request.spi.info.family = PF_INET; + request.spi.min = 0xc0000000; + request.spi.max = 0xcFFFFFFF; + + if (this->send_message(this, &request, &response) != SUCCESS) + { + this->logger->log(this->logger, ERROR, "netlink communication failed"); + return FAILED; + } + else if (response->hdr.nlmsg_type == NLMSG_ERROR) + { + this->logger->log(this->logger, ERROR, "netlink request XFRM_MSG_ALLOCSPI got an error: %s", + strerror(-response->e.error)); + status = FAILED; + } + else if (response->hdr.nlmsg_type != XFRM_MSG_NEWSA) + { + this->logger->log(this->logger, ERROR, "netlink request XFRM_MSG_ALLOCSPI got a unknown reply"); + status = FAILED; + } + else if (response->hdr.nlmsg_len < NLMSG_LENGTH(sizeof(response->sa))) + { + this->logger->log(this->logger, ERROR, "netlink request XFRM_MSG_ALLOCSPI got an invalid reply"); + status = FAILED; + } + else + { + *spi = response->sa.id.spi; + } + free(response); + + return status; +} + +/** + * Implementation of kernel_interface_t.add_sa. + */ +static status_t add_sa( private_kernel_interface_t *this, + host_t *me, + host_t *other, + u_int32_t spi, + int protocol, + u_int32_t reqid, + encryption_algorithm_t enc_alg, + chunk_t encryption_key, + integrity_algorithm_t int_alg, + chunk_t integrity_key, + bool replace) +{ + netlink_message_t request, *response; + memset(&request, 0, sizeof(request)); + status_t status = SUCCESS; + + this->logger->log(this->logger, CONTROL|LEVEL2, "adding SA"); + + request.hdr.nlmsg_flags = NLM_F_REQUEST | NLM_F_ACK; + request.hdr.nlmsg_type = replace ? XFRM_MSG_UPDSA : XFRM_MSG_NEWSA; + + request.sa.saddr = me->get_xfrm_addr(me); + request.sa.id.daddr = other->get_xfrm_addr(other); + + request.sa.id.spi = spi; + request.sa.id.proto = (protocol == PROTO_ESP) ? KERNEL_ESP : KERNEL_AH; + request.sa.family = me->get_family(me); + request.sa.mode = TRUE; /* tunnel mode */ + request.sa.replay_window = 32; + request.sa.reqid = reqid; + request.sa.lft.soft_byte_limit = XFRM_INF; + request.sa.lft.soft_packet_limit = XFRM_INF; + request.sa.lft.hard_byte_limit = XFRM_INF; + request.sa.lft.hard_packet_limit = XFRM_INF; + + request.hdr.nlmsg_len = NLMSG_ALIGN(NLMSG_LENGTH(sizeof(request.sa))); + + if (enc_alg != ENCR_UNDEFINED) + { + xfrm_data_t *data = (xfrm_data_t*)(((u_int8_t*)&request) + request.hdr.nlmsg_len); + + data->type = XFRMA_ALG_CRYPT; + data->length = 4 + sizeof(data->algo) + encryption_key.len; + data->algo.alg_key_len = encryption_key.len * 8; + request.hdr.nlmsg_len += data->length; + if (request.hdr.nlmsg_len > sizeof(request)) + { + return FAILED; + } + strcpy(data->algo.alg_name, mapping_find(kernel_encryption_algs_m, enc_alg)); + memcpy(data->algo.alg_key, encryption_key.ptr, encryption_key.len); + } + + if (int_alg != AUTH_UNDEFINED) + { + xfrm_data_t *data = (xfrm_data_t*)(((u_int8_t*)&request) + request.hdr.nlmsg_len); + + data->type = XFRMA_ALG_AUTH; + data->length = 4 + sizeof(data->algo) + integrity_key.len; + data->algo.alg_key_len = integrity_key.len * 8; + request.hdr.nlmsg_len += data->length; + if (request.hdr.nlmsg_len > sizeof(request)) + { + return FAILED; + } + strcpy(data->algo.alg_name, mapping_find(kernel_integrity_algs_m, int_alg)); + memcpy(data->algo.alg_key, integrity_key.ptr, integrity_key.len); + } + + /* TODO: add IPComp here*/ + + if (this->send_message(this, &request, &response) != SUCCESS) + { + this->logger->log(this->logger, ERROR, "netlink communication failed"); + return FAILED; + } + else if (response->hdr.nlmsg_type != NLMSG_ERROR) + { + this->logger->log(this->logger, ERROR, "netlink request XFRM_MSG_NEWSA not acknowledged"); + status = FAILED; + } + else if (response->e.error) + { + this->logger->log(this->logger, ERROR, "netlink request XFRM_MSG_NEWSA got error %s", + strerror(-response->e.error)); + status = FAILED; + } + + free(response); + return status; +} + +static status_t del_sa( private_kernel_interface_t *this, + host_t *dst, + u_int32_t spi, + protocol_id_t protocol) +{ + netlink_message_t request, *response; + memset(&request, 0, sizeof(request)); + status_t status = SUCCESS; + + this->logger->log(this->logger, CONTROL|LEVEL2, "deleting SA"); + + request.hdr.nlmsg_flags = NLM_F_REQUEST | NLM_F_ACK; + request.hdr.nlmsg_type = XFRM_MSG_DELSA; + + request.sa_id.daddr = dst->get_xfrm_addr(dst); + + request.sa_id.spi = spi; + request.sa_id.proto = (protocol == PROTO_ESP) ? KERNEL_ESP : KERNEL_AH; + request.sa_id.family = dst->get_family(dst); + + request.hdr.nlmsg_len = NLMSG_ALIGN(NLMSG_LENGTH(sizeof(request.sa_id))); + + if (this->send_message(this, &request, &response) != SUCCESS) + { + return FAILED; + } + else if (response->hdr.nlmsg_type != NLMSG_ERROR) + { + status = FAILED; + } + else if (response->e.error) + { + status = FAILED; + } + + free(response); + return status; +} + +/** + * Implementation of kernel_interface_t.add_policy. + */ +static status_t add_policy(private_kernel_interface_t *this, + host_t *me, host_t *other, + host_t *src, host_t *dst, + u_int8_t src_hostbits, u_int8_t dst_hostbits, + int direction, int upper_proto, + bool ah, bool esp, + u_int32_t reqid) +{ + netlink_message_t request, *response; + status_t status = SUCCESS; + + this->logger->log(this->logger, CONTROL|LEVEL2, "adding policy"); + + memset(&request, 0, sizeof(request)); + request.hdr.nlmsg_flags = NLM_F_REQUEST | NLM_F_ACK; + + request.policy.sel.sport = htons(src->get_port(src)); + request.policy.sel.dport = htons(dst->get_port(dst)); + request.policy.sel.sport_mask = (request.policy.sel.sport) ? ~0 : 0; + request.policy.sel.dport_mask = (request.policy.sel.dport) ? ~0 : 0; + request.policy.sel.saddr = src->get_xfrm_addr(src); + request.policy.sel.daddr = dst->get_xfrm_addr(dst); + request.policy.sel.prefixlen_s = src_hostbits; + request.policy.sel.prefixlen_d = dst_hostbits; + request.policy.sel.proto = upper_proto; + request.policy.sel.family = src->get_family(src); + + request.hdr.nlmsg_type = XFRM_MSG_NEWPOLICY; + request.hdr.nlmsg_len = NLMSG_ALIGN(NLMSG_LENGTH(sizeof(request.policy))); + + request.policy.dir = direction; + request.policy.priority = SPD_PRIORITY; + request.policy.action = XFRM_POLICY_ALLOW; + request.policy.share = XFRM_SHARE_ANY; + + request.policy.lft.soft_byte_limit = XFRM_INF; + request.policy.lft.soft_packet_limit = XFRM_INF; + request.policy.lft.hard_byte_limit = XFRM_INF; + request.policy.lft.hard_packet_limit = XFRM_INF; + + if (esp || ah) + { + xfrm_data_t *data; + int tmpl_pos = 0; + data = (xfrm_data_t*)(((u_int8_t*)&request) + request.hdr.nlmsg_len); + data->type = XFRMA_TMPL; + if (esp) + { + data->tmpl[tmpl_pos].reqid = reqid; + data->tmpl[tmpl_pos].id.proto = KERNEL_ESP; + data->tmpl[tmpl_pos].aalgos = data->tmpl[tmpl_pos].ealgos = data->tmpl[tmpl_pos].calgos = ~0; + data->tmpl[tmpl_pos].mode = TRUE; + + data->tmpl[tmpl_pos].saddr = me->get_xfrm_addr(me); + data->tmpl[tmpl_pos].id.daddr = me->get_xfrm_addr(other); + + tmpl_pos++; + } + if (ah) + { + data->tmpl[tmpl_pos].reqid = reqid; + data->tmpl[tmpl_pos].id.proto = KERNEL_AH; + data->tmpl[tmpl_pos].aalgos = data->tmpl[tmpl_pos].ealgos = data->tmpl[tmpl_pos].calgos = ~0; + data->tmpl[tmpl_pos].mode = TRUE; + + data->tmpl[tmpl_pos].saddr = me->get_xfrm_addr(me); + data->tmpl[tmpl_pos].id.daddr = other->get_xfrm_addr(other); + + tmpl_pos++; + } + data->length = 4 + sizeof(struct xfrm_user_tmpl) * tmpl_pos; + request.hdr.nlmsg_len += data->length; + } + + if (this->send_message(this, &request, &response) != SUCCESS) + { + this->logger->log(this->logger, ERROR, "netlink communication failed"); + return FAILED; + } + else if (response->hdr.nlmsg_type != NLMSG_ERROR) + { + this->logger->log(this->logger, ERROR, "netlink request XFRM_MSG_NEWPOLICY not acknowledged"); + status = FAILED; + } + else if (response->e.error) + { + this->logger->log(this->logger, ERROR, "netlink request XFRM_MSG_NEWPOLICY got error %s", + strerror(-response->e.error)); + status = FAILED; + } + + free(response); + return status; +} + +/** + * Implementation of kernel_interface_t.del_policy. + */ +static status_t del_policy(private_kernel_interface_t *this, + host_t *me, host_t *other, + host_t *src, host_t *dst, + u_int8_t src_hostbits, u_int8_t dst_hostbits, + int direction, int upper_proto) +{ + netlink_message_t request, *response; + status_t status = SUCCESS; + + + this->logger->log(this->logger, CONTROL|LEVEL2, "deleting policy"); + + memset(&request, 0, sizeof(request)); + request.hdr.nlmsg_flags = NLM_F_REQUEST | NLM_F_ACK; + + request.policy_id.sel.sport = htons(src->get_port(src)); + request.policy_id.sel.dport = htons(dst->get_port(dst)); + request.policy_id.sel.sport_mask = (request.policy.sel.sport) ? ~0 : 0; + request.policy_id.sel.dport_mask = (request.policy.sel.dport) ? ~0 : 0; + request.policy_id.sel.saddr = src->get_xfrm_addr(src); + request.policy_id.sel.daddr = dst->get_xfrm_addr(dst); + request.policy_id.sel.prefixlen_s = src_hostbits; + request.policy_id.sel.prefixlen_d = dst_hostbits; + request.policy_id.sel.proto = upper_proto; + request.policy_id.sel.family = src->get_family(src); + + request.policy_id.dir = direction; + + request.hdr.nlmsg_type = XFRM_MSG_DELPOLICY; + request.hdr.nlmsg_len = NLMSG_ALIGN(NLMSG_LENGTH(sizeof(request.policy_id))); + + if (this->send_message(this, &request, &response) != SUCCESS) + { + return FAILED; + } + else if (response->hdr.nlmsg_type != NLMSG_ERROR) + { + status = FAILED; + } + else if (response->e.error) + { + status = FAILED; + } + + free(response); + return status; +} + +/** + * Implementation of private_kernel_interface_t.send_message. + */ +static status_t send_message(private_kernel_interface_t *this, netlink_message_t *request, netlink_message_t **response) +{ + size_t length; + struct sockaddr_nl addr; + + request->hdr.nlmsg_seq = ++this->seq; + request->hdr.nlmsg_pid = this->pid; + + memset(&addr, 0, sizeof(struct sockaddr_nl)); + addr.nl_family = AF_NETLINK; + addr.nl_pid = 0; + addr.nl_groups = 0; + + length = sendto(this->socket,(void *)request, request->hdr.nlmsg_len, 0, (struct sockaddr *)&addr, sizeof(addr)); + + if (length < 0) + { + return FAILED; + } + else if (length != request->hdr.nlmsg_len) + { + return FAILED; + } + + pthread_mutex_lock(&(this->mutex)); + + while (TRUE) + { + iterator_t *iterator; + bool found = FALSE; + /* search list, break if found */ + iterator = this->responses->create_iterator(this->responses, TRUE); + while (iterator->has_next(iterator)) + { + netlink_message_t *listed_response; + iterator->current(iterator, (void**)&listed_response); + if (listed_response->hdr.nlmsg_seq == request->hdr.nlmsg_seq) + { + /* matches our request, this is the reply */ + *response = listed_response; + found = TRUE; + break; + } + } + iterator->destroy(iterator); + + if (found) + { + break; + } + /* TODO: we should time out, if something goes wrong!??? */ + pthread_cond_wait(&(this->condvar), &(this->mutex)); + } + + pthread_mutex_unlock(&(this->mutex)); + + return SUCCESS; +} + +/** + * Implementation of private_kernel_interface_t.receive_messages. + */ +static void receive_messages(private_kernel_interface_t *this) +{ + while(TRUE) + { + netlink_message_t response, *listed_response; + while (TRUE) + { + struct sockaddr_nl addr; + socklen_t addr_length; + size_t length; + + addr_length = sizeof(addr); + + response.hdr.nlmsg_type = XFRM_MSG_NEWSA; + length = recvfrom(this->socket, &response, sizeof(response), 0, (struct sockaddr*)&addr, &addr_length); + if (length < 0) + { + if (errno == EINTR) + { + /* interrupted, try again */ + continue; + } + charon->kill(charon, "receiving from netlink socket failed"); + } + if (!NLMSG_OK(&response.hdr, length)) + { + /* bad netlink message */ + continue; + } + if (addr.nl_pid != 0) + { + /* not from kernel. not interested, try another one */ + continue; + } + break; + } + + /* got a valid message. + * requests are handled on our own, + * responses are listed for the requesters + */ + if (response.hdr.nlmsg_flags & NLM_F_REQUEST) + { + /* handle request */ + } + else + { + /* add response to queue */ + listed_response = malloc(sizeof(response)); + memcpy(listed_response, &response, sizeof(response)); + + pthread_mutex_lock(&(this->mutex)); + this->responses->insert_last(this->responses, (void*)listed_response); + pthread_mutex_unlock(&(this->mutex)); + /* signal ALL waiting threads */ + pthread_cond_broadcast(&(this->condvar)); + } + /* get the next one */ + } +} + +/** + * Implementation of kernel_interface_t.destroy. + */ +static void destroy(private_kernel_interface_t *this) +{ + pthread_cancel(this->thread); + pthread_join(this->thread, NULL); + close(this->socket); + this->responses->destroy(this->responses); + free(this); +} + +/* + * Described in header. + */ +kernel_interface_t *kernel_interface_create() +{ + private_kernel_interface_t *this = malloc_thing(private_kernel_interface_t); + + /* public functions */ + this->public.get_spi = (status_t(*)(kernel_interface_t*,host_t*,host_t*,protocol_id_t,u_int32_t,u_int32_t*))get_spi; + this->public.add_sa = (status_t(*)(kernel_interface_t *,host_t*,host_t*,u_int32_t,protocol_id_t,u_int32_t,encryption_algorithm_t,chunk_t,integrity_algorithm_t,chunk_t,bool))add_sa; + this->public.add_policy = (status_t(*)(kernel_interface_t*,host_t*, host_t*,host_t*,host_t*,u_int8_t,u_int8_t,int,int,bool,bool,u_int32_t))add_policy; + this->public.del_sa = (status_t(*)(kernel_interface_t*,host_t*,u_int32_t,protocol_id_t))del_sa; + this->public.del_policy = (status_t(*)(kernel_interface_t*,host_t*,host_t*,host_t*,host_t*,u_int8_t,u_int8_t,int,int))del_policy; + + this->public.destroy = (void(*)(kernel_interface_t*)) destroy; + + /* private members */ + this->receive_messages = receive_messages; + this->send_message = send_message; + this->pid = getpid(); + this->responses = linked_list_create(); + this->logger = logger_manager->get_logger(logger_manager, XFRM); + pthread_mutex_init(&(this->mutex),NULL); + pthread_cond_init(&(this->condvar),NULL); + this->seq = 0; + this->socket = socket(PF_NETLINK, SOCK_RAW, NETLINK_XFRM); + if (this->socket <= 0) + { + this->responses->destroy(this->responses); + free(this); + charon->kill(charon, "Unable to create netlink socket"); + } + + if (pthread_create(&(this->thread), NULL, (void*(*)(void*))this->receive_messages, this) != 0) + { + this->responses->destroy(this->responses); + close(this->socket); + free(this); + charon->kill(charon, "Unable to create netlink thread"); + } + + return (&this->public); +} diff --git a/src/charon/threads/kernel_interface.h b/src/charon/threads/kernel_interface.h new file mode 100644 index 000000000..b3ca13faa --- /dev/null +++ b/src/charon/threads/kernel_interface.h @@ -0,0 +1,185 @@ +/** + * @file kernel_interface.h + * + * @brief Interface of kernel_interface_t. + * + */ + +/* + * Copyright (C) 2005 Jan Hutter, Martin Willi + * Hochschule fuer Technik Rapperswil + * + * This program is free software; you can redistribute it and/or modify it + * under the terms of the GNU General Public License as published by the + * Free Software Foundation; either version 2 of the License, or (at your + * option) any later version. See <http://www.fsf.org/copyleft/gpl.txt>. + * + * 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 General Public License + * for more details. + */ + +#ifndef KERNEL_INTERFACE_H_ +#define KERNEL_INTERFACE_H_ + +#include <linux/xfrm.h> + +#include <utils/host.h> +#include <encoding/payloads/proposal_substructure.h> + +typedef struct kernel_interface_t kernel_interface_t; + +/** + * @brief Interface to the kernel. + * + * The kernel interface handles the communication with the kernel + * for SA and policy management. It allows setup of these, and provides + * further the handling of kernel events. + * + * @b Constructors: + * - kernel_interface_create() + * + * @ingroup threads + */ +struct kernel_interface_t { + + /** + * @brief Get a SPI from the kernel. + * + * @param this calling object + * @param src source address of SA + * @param dst destination address of SA + * @param protocol protocol for SA (ESP/AH) + * @param reqid unique ID for this SA + * @param[out] spi allocated spi + * @return + * - SUCCESS + * - FAILED if kernel comm failed + */ + status_t (*get_spi) (kernel_interface_t *this, + host_t *src, host_t *dst, + protocol_id_t protocol, + u_int32_t reqid, + u_int32_t *spi); + + /** + * @brief Add an SA to the SAD. + * + * add_sa() may update an already allocated + * SPI (via get_spi). In this case, the replace + * flag must be set. + * This function does install a single SA for a + * single protocol in one direction. + * + * @param this calling object + * @param src source address for this SA + * @param dst destination address for this SA + * @param spi SPI allocated by us or remote peer + * @param protocol protocol for this SA (ESP/AH) + * @param reqid unique ID for this SA + * @param enc_alg Algorithm to use for encryption (ESP only) + * @param enc_key Key to use for encryption + * @param int_alg Algorithm to use for integrity protection + * @param int_key Key for integrity protection + * @param replace Should an already installed SA be updated? + * @return + * - SUCCESS + * - FAILED if kernel comm failed + */ + status_t (*add_sa)(kernel_interface_t *this, + host_t *src, host_t *dst, + u_int32_t spi, + protocol_id_t protocol, + u_int32_t reqid, + encryption_algorithm_t enc_alg, + chunk_t enc_key, + integrity_algorithm_t int_alg, + chunk_t int_key, + bool replace); + /** + * @brief Delete a previusly installed SA from the SAD. + * + * @param this calling object + * @param dst destination address for this SA + * @param spi SPI allocated by us or remote peer + * @param protocol protocol for this SA (ESP/AH) + * @return + * - SUCCESS + * - FAILED if kernel comm failed + */ + status_t (*del_sa) (kernel_interface_t *this, + host_t *dst, + u_int32_t spi, + protocol_id_t protocol); + + /** + * @brief Add a policy to the SPD. + * + * A policy is always associated to an SA, so + * traffic applied to a policy. Traffic which + * matches a policy is handled by the SA with the same + * reqid. + * + * @param this calling object + * @param me address of local peer + * @param other address of remote peer + * @param src src address of traffic this policy applies + * @param dst dest address of traffic this policy applies + * @param src_hostbits subnetmask to use for src address + * @param dst_hostbits subnetmask to use for dst address + * @param direction direction of traffic, XFRM_POLICY_OUT, XFRM_POLICY_IN, XFRM_POLICY_FWD + * @param upper_proto upper layer protocol of traffic for this policy (TCP, UDP, ICMP, ...) + * @param ah protect traffic with AH? + * @param esp protect traffic with ESP? + * @param reqid uniqe ID of an SA to use to enforce policy + * @return + * - SUCCESS + * - FAILED if kernel comm failed + */ + status_t (*add_policy) (kernel_interface_t *this, + host_t *me, host_t *other, + host_t *src, host_t *dst, + u_int8_t src_hostbits, u_int8_t dst_hostbits, + int direction, int upper_proto, + bool ah, bool esp, + u_int32_t reqid); + + /** + * @brief Remove a policy from the SPD. + * + * @param this calling object + * @param me address of local peer + * @param other address of remote peer + * @param src src address of traffic this policy applies + * @param dst dest address of traffic this policy applies + * @param src_hostbits subnetmask to use for src address + * @param dst_hostbits subnetmask to use for dst address + * @param direction direction of traffic, XFRM_POLICY_OUT, XFRM_POLICY_IN, XFRM_POLICY_FWD + * @param upper_proto upper layer protocol of traffic for this policy (TCP, UDP, ICMP, ...) + * @return + * - SUCCESS + * - FAILED if kernel comm failed + */ + status_t (*del_policy) (kernel_interface_t *this, + host_t *me, host_t *other, + host_t *src, host_t *dst, + u_int8_t src_hostbits, u_int8_t dst_hostbits, + int direction, int upper_proto); + + /** + * @brief Destroys a kernel_interface object. + * + * @param kernel_interface_t calling object + */ + void (*destroy) (kernel_interface_t *kernel_interface); +}; + +/** + * @brief Creates an object of type kernel_interface_t. + * + * @ingroup threads + */ +kernel_interface_t *kernel_interface_create(void); + +#endif /*KERNEL_INTERFACE_H_*/ diff --git a/src/charon/threads/receiver.c b/src/charon/threads/receiver.c new file mode 100644 index 000000000..0cf8b7bde --- /dev/null +++ b/src/charon/threads/receiver.c @@ -0,0 +1,128 @@ +/** + * @file receiver.c + * + * @brief Implementation of receiver_t. + * + */ + +/* + * Copyright (C) 2005 Jan Hutter, Martin Willi + * Hochschule fuer Technik Rapperswil + * + * This program is free software; you can redistribute it and/or modify it + * under the terms of the GNU General Public License as published by the + * Free Software Foundation; either version 2 of the License, or (at your + * option) any later version. See <http://www.fsf.org/copyleft/gpl.txt>. + * + * 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 General Public License + * for more details. + */ + +#include <stdlib.h> +#include <pthread.h> + +#include "receiver.h" + +#include <daemon.h> +#include <network/socket.h> +#include <network/packet.h> +#include <queues/job_queue.h> +#include <queues/jobs/job.h> +#include <queues/jobs/incoming_packet_job.h> +#include <utils/logger_manager.h> + + +typedef struct private_receiver_t private_receiver_t; + +/** + * Private data of a receiver_t object. + */ +struct private_receiver_t { + /** + * Public part of a receiver_t object. + */ + receiver_t public; + + /** + * @brief Thread function started at creation of the receiver object. + * + * @param this calling object + */ + void (*receive_packets) (private_receiver_t *this); + + /** + * Assigned thread. + */ + pthread_t assigned_thread; + + /** + * A logger for the receiver_t object. + */ + logger_t *logger; +}; + +/** + * Implementation of receiver_t.receive_packets. + */ +static void receive_packets(private_receiver_t * this) +{ + packet_t * current_packet; + job_t *current_job; + + /* cancellation disabled by default */ + pthread_setcancelstate(PTHREAD_CANCEL_DISABLE, NULL); + + this->logger->log(this->logger, CONTROL, "Receiver thread running, thread_id %u", (int)pthread_self()); + + while (1) + { + while (charon->socket->receive(charon->socket,¤t_packet) == SUCCESS) + { + this->logger->log(this->logger, CONTROL | LEVEL1, "Creating job from packet"); + current_job = (job_t *) incoming_packet_job_create(current_packet); + + charon->job_queue->add(charon->job_queue,current_job); + + } + /* bad bad, rebuild the socket ? */ + this->logger->log(this->logger, ERROR, "Receiving from socket failed!"); + } +} + +/** + * Implementation of receiver_t.destroy. + */ +static void destroy(private_receiver_t *this) +{ + this->logger->log(this->logger, CONTROL | LEVEL1, "Going to terminate receiver thread"); + pthread_cancel(this->assigned_thread); + + pthread_join(this->assigned_thread, NULL); + this->logger->log(this->logger, CONTROL | LEVEL1, "Receiver thread terminated"); + + free(this); +} + +/* + * Described in header. + */ +receiver_t * receiver_create() +{ + private_receiver_t *this = malloc_thing(private_receiver_t); + + this->public.destroy = (void(*)(receiver_t*)) destroy; + this->receive_packets = receive_packets; + + this->logger = logger_manager->get_logger(logger_manager, RECEIVER); + + if (pthread_create(&(this->assigned_thread), NULL, (void*(*)(void*))this->receive_packets, this) != 0) + { + this->logger->log(this->logger, ERROR, "Receiver thread could not be started"); + free(this); + charon->kill(charon, "Unable to create receiver thread"); + } + + return &(this->public); +} diff --git a/src/charon/threads/receiver.h b/src/charon/threads/receiver.h new file mode 100644 index 000000000..1f9b644b1 --- /dev/null +++ b/src/charon/threads/receiver.h @@ -0,0 +1,67 @@ +/** + * @file receiver.h + * + * @brief Interface of receiver_t. + * + */ + +/* + * Copyright (C) 2005 Jan Hutter, Martin Willi + * Hochschule fuer Technik Rapperswil + * + * This program is free software; you can redistribute it and/or modify it + * under the terms of the GNU General Public License as published by the + * Free Software Foundation; either version 2 of the License, or (at your + * option) any later version. See <http://www.fsf.org/copyleft/gpl.txt>. + * + * 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 General Public License + * for more details. + */ + +#ifndef RECEIVER_H_ +#define RECEIVER_H_ + +#include <types.h> + + +typedef struct receiver_t receiver_t; + +/** + * @brief Receives packets from the socket and adds them to the job queue. + * + * The receiver starts a thread, wich reads on the blocking socket. If + * data is available, a packet_t object is created , wrapped + * in an incoming_packet_job_t and added to the job queue. + * + * @b Constructors: + * - receiver_create() + * + * @ingroup threads + */ +struct receiver_t { + + /** + * @brief Destroys a receiver_t object. + * + * @param receiver receiver object + */ + void (*destroy) (receiver_t *receiver); +}; + +/** + * @brief Create a receiver_t object. + * + * The receiver thread will start working, get data + * from the socket and add those packets to the job queue. + * + * @return + * - receiver_t object + * - NULL of thread could not be started + * + * @ingroup threads + */ +receiver_t * receiver_create(void); + +#endif /*RECEIVER_H_*/ diff --git a/src/charon/threads/scheduler.c b/src/charon/threads/scheduler.c new file mode 100644 index 000000000..47c5d6fb9 --- /dev/null +++ b/src/charon/threads/scheduler.c @@ -0,0 +1,124 @@ +/** + * @file scheduler.c + * + * @brief Implementation of scheduler_t. + * + */ + +/* + * Copyright (C) 2005 Jan Hutter, Martin Willi + * Hochschule fuer Technik Rapperswil + * + * This program is free software; you can redistribute it and/or modify it + * under the terms of the GNU General Public License as published by the + * Free Software Foundation; either version 2 of the License, or (at your + * option) any later version. See <http://www.fsf.org/copyleft/gpl.txt>. + * + * 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 General Public License + * for more details. + */ + +#include <stdlib.h> +#include <pthread.h> + +#include "scheduler.h" + +#include <daemon.h> +#include <definitions.h> +#include <utils/logger_manager.h> +#include <queues/job_queue.h> + + +typedef struct private_scheduler_t private_scheduler_t; + +/** + * Private data of a scheduler_t object. + */ +struct private_scheduler_t { + /** + * Public part of a scheduler_t object. + */ + scheduler_t public; + + /** + * @brief Get events from the event queue and add them to to job queue. + * + * Thread function started at creation of the scheduler object. + * + * @param this calling object + */ + void (*get_events) (private_scheduler_t *this); + + /** + * Assigned thread. + */ + pthread_t assigned_thread; + + /** + * A logger. + */ + logger_t *logger; +}; + +/** + * Implementation of private_scheduler_t.get_events. + */ +static void get_events(private_scheduler_t * this) +{ + job_t *current_job; + + /* cancellation disabled by default */ + pthread_setcancelstate(PTHREAD_CANCEL_DISABLE, NULL); + + this->logger->log(this->logger, CONTROL, "Scheduler thread running, thread_id %u", (int)pthread_self()); + + for (;;) + { + this->logger->log(this->logger, CONTROL|LEVEL2, "Waiting for next event..."); + /* get a job, this block until one is available */ + current_job = charon->event_queue->get(charon->event_queue); + /* queue the job in the job queue, workers will eat them */ + charon->job_queue->add(charon->job_queue, current_job); + this->logger->log(this->logger, CONTROL | LEVEL1, "Got event, added job %s to job-queue.", + mapping_find(job_type_m, current_job->get_type(current_job))); + } +} + +/** + * Implementation of scheduler_t.destroy. + */ +static void destroy(private_scheduler_t *this) +{ + this->logger->log(this->logger, CONTROL | LEVEL1, "Going to terminate scheduler thread"); + pthread_cancel(this->assigned_thread); + + pthread_join(this->assigned_thread, NULL); + this->logger->log(this->logger, CONTROL | LEVEL1, "Scheduler thread terminated"); + + free(this); +} + +/* + * Described in header. + */ +scheduler_t * scheduler_create() +{ + private_scheduler_t *this = malloc_thing(private_scheduler_t); + + this->public.destroy = (void(*)(scheduler_t*)) destroy; + this->get_events = get_events; + + this->logger = logger_manager->get_logger(logger_manager, SCHEDULER); + + if (pthread_create(&(this->assigned_thread), NULL, (void*(*)(void*))this->get_events, this) != 0) + { + /* thread could not be created */ + this->logger->log(this->logger, ERROR, "Scheduler thread could not be created!"); + free(this); + charon->kill(charon, "Unable to create scheduler thread"); + } + + return &(this->public); +} diff --git a/src/charon/threads/scheduler.h b/src/charon/threads/scheduler.h new file mode 100644 index 000000000..04d639d1b --- /dev/null +++ b/src/charon/threads/scheduler.h @@ -0,0 +1,67 @@ +/** + * @file scheduler.h + * + * @brief Interface of scheduler_t. + * + */ + +/* + * Copyright (C) 2005 Jan Hutter, Martin Willi + * Hochschule fuer Technik Rapperswil + * + * This program is free software; you can redistribute it and/or modify it + * under the terms of the GNU General Public License as published by the + * Free Software Foundation; either version 2 of the License, or (at your + * option) any later version. See <http://www.fsf.org/copyleft/gpl.txt>. + * + * 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 General Public License + * for more details. + */ + +#ifndef SCHEDULER_H_ +#define SCHEDULER_H_ + +#include <types.h> + +typedef struct scheduler_t scheduler_t; + +/** + * @brief The scheduler thread is responsible for timed events. + * + * The scheduler thread takes out jobs from the event-queue and adds them + * to the job-queue. + * + * Starts a thread which does the work, since event-queue is blocking. + * + * @b Constructors: + * - scheduler_create() + * + * @ingroup threads + */ +struct scheduler_t { + + /** + * @brief Destroys a scheduler object. + * + * @param scheduler calling object + */ + void (*destroy) (scheduler_t *scheduler); +}; + +/** + * @brief Create a scheduler with its associated thread. + * + * The thread will start to get jobs form the event queue + * and adds them to the job queue. + * + * @return + * - scheduler_t object + * - NULL if thread could not be started + * + * @ingroup threads + */ +scheduler_t * scheduler_create(void); + +#endif /*SCHEDULER_H_*/ diff --git a/src/charon/threads/sender.c b/src/charon/threads/sender.c new file mode 100644 index 000000000..42d11beb9 --- /dev/null +++ b/src/charon/threads/sender.c @@ -0,0 +1,126 @@ +/** + * @file sender.c + * + * @brief Implementation of sender_t. + * + */ + +/* + * Copyright (C) 2005 Jan Hutter, Martin Willi + * Hochschule fuer Technik Rapperswil + * + * This program is free software; you can redistribute it and/or modify it + * under the terms of the GNU General Public License as published by the + * Free Software Foundation; either version 2 of the License, or (at your + * option) any later version. See <http://www.fsf.org/copyleft/gpl.txt>. + * + * 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 General Public License + * for more details. + */ + +#include <stdlib.h> +#include <pthread.h> + +#include "sender.h" + +#include <daemon.h> +#include <network/socket.h> +#include <network/packet.h> +#include <queues/send_queue.h> +#include <utils/logger_manager.h> + + +typedef struct private_sender_t private_sender_t; + +/** + * Private data of a sender_t object. + */ +struct private_sender_t { + /** + * Public part of a sender_t object. + */ + sender_t public; + + /** + * Assigned thread. + */ + pthread_t assigned_thread; + + /** + * @brief The thread function, sends out packets. + * + * @param this calling object + */ + void (*send_packets) (private_sender_t * this); + + /** + * A logger for this sender_t object. + */ + logger_t *logger; + +}; + +/** + * Implementation of private_sender_t.send_packets. + */ +static void send_packets(private_sender_t * this) +{ + packet_t * current_packet; + status_t status; + + /* cancellation disabled by default */ + pthread_setcancelstate(PTHREAD_CANCEL_DISABLE, NULL); + + this->logger->log(this->logger, CONTROL, "Sender thread running, thread_id %u", (int)pthread_self()); + + while (1) + { + current_packet = charon->send_queue->get(charon->send_queue); + this->logger->log(this->logger, CONTROL|LEVEL1, "Got a packet, sending it"); + status = charon->socket->send(charon->socket,current_packet); + if (status != SUCCESS) + { + this->logger->log(this->logger, ERROR, "Sending failed, socket returned %s", + mapping_find(status_m, status)); + } + current_packet->destroy(current_packet); + } +} + +/** + * Implementation of sender_t.destroy. + */ +static void destroy(private_sender_t *this) +{ + this->logger->log(this->logger, CONTROL | LEVEL1, "Going to terminate sender thread"); + pthread_cancel(this->assigned_thread); + + pthread_join(this->assigned_thread, NULL); + this->logger->log(this->logger, CONTROL | LEVEL1, "Sender thread terminated"); + + free(this); +} + +/* + * Described in header. + */ +sender_t * sender_create() +{ + private_sender_t *this = malloc_thing(private_sender_t); + + this->send_packets = send_packets; + this->public.destroy = (void(*)(sender_t*)) destroy; + + this->logger = logger_manager->get_logger(logger_manager, SENDER); + + if (pthread_create(&(this->assigned_thread), NULL, (void*(*)(void*))this->send_packets, this) != 0) + { + this->logger->log(this->logger, ERROR, "Sender thread could not be created"); + free(this); + charon->kill(charon, "Unable to create sender thread"); + } + + return &(this->public); +} diff --git a/src/charon/threads/sender.h b/src/charon/threads/sender.h new file mode 100644 index 000000000..2a41ccc66 --- /dev/null +++ b/src/charon/threads/sender.h @@ -0,0 +1,63 @@ +/** + * @file sender.h + * + * @brief Interface of sender_t. + * + */ + +/* + * Copyright (C) 2005 Jan Hutter, Martin Willi + * Hochschule fuer Technik Rapperswil + * + * This program is free software; you can redistribute it and/or modify it + * under the terms of the GNU General Public License as published by the + * Free Software Foundation; either version 2 of the License, or (at your + * option) any later version. See <http://www.fsf.org/copyleft/gpl.txt>. + * + * 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 General Public License + * for more details. + */ + +#ifndef SENDER_H_ +#define SENDER_H_ + +#include <types.h> + +typedef struct sender_t sender_t; + +/** + * @brief Thread responsible for sending packets over the socket. + * + * @b Constructors: + * - sender_create() + * + * @ingroup threads + */ +struct sender_t { + + /** + * @brief Destroys a sender object. + * + * @param sender calling object + */ + void (*destroy) (sender_t *sender); +}; + + +/** + * @brief Create the sender thread. + * + * The thread will start to work, getting packets + * from the send queue and sends them out. + * + * @return + * - sender_t object + * - NULL of thread could not be started + * + * @ingroup threads + */ +sender_t * sender_create(void); + +#endif /*SENDER_H_*/ diff --git a/src/charon/threads/stroke_interface.c b/src/charon/threads/stroke_interface.c new file mode 100755 index 000000000..a9c55673b --- /dev/null +++ b/src/charon/threads/stroke_interface.c @@ -0,0 +1,646 @@ +/** + * @file stroke.c + * + * @brief Implementation of stroke_t. + * + */ + +/* + * Copyright (C) 2006 Martin Willi + * Hochschule fuer Technik Rapperswil + * + * This program is free software; you can redistribute it and/or modify it + * under the terms of the GNU General Public License as published by the + * Free Software Foundation; either version 2 of the License, or (at your + * option) any later version. See <http://www.fsf.org/copyleft/gpl.txt>. + * + * 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 General Public License + * for more details. + */ + +#include <stdlib.h> +#include <sys/types.h> +#include <sys/stat.h> +#include <sys/socket.h> +#include <sys/un.h> +#include <sys/fcntl.h> +#include <unistd.h> +#include <dirent.h> +#include <errno.h> +#include <pthread.h> + +#include "stroke_interface.h" + +#include <stroke.h> +#include <types.h> +#include <daemon.h> +#include <crypto/x509.h> +#include <queues/jobs/initiate_ike_sa_job.h> + + +struct sockaddr_un socket_addr = { AF_UNIX, STROKE_SOCKET}; + + +typedef struct private_stroke_t private_stroke_t; + +/** + * Private data of an stroke_t object. + */ +struct private_stroke_t { + + /** + * Public part of stroke_t object. + */ + stroke_t public; + + /** + * Assigned logger_t object in charon. + */ + logger_t *logger; + + /** + * Logger which logs to stroke + */ + logger_t *stroke_logger; + + /** + * Unix socket to listen for strokes + */ + int socket; + + /** + * Thread which reads from the socket + */ + pthread_t assigned_thread; + + /** + * Read from the socket and handle stroke messages + */ + void (*stroke_receive) (private_stroke_t *this); +}; + +/** + * Helper function which corrects the string pointers + * in a stroke_msg_t. Strings in a stroke_msg sent over "wire" + * contains RELATIVE addresses (relative to the beginning of the + * stroke_msg). They must be corrected if they reach our address + * space... + */ +static void pop_string(stroke_msg_t *msg, char **string) +{ + /* check for sanity of string pointer and string */ + if (*string == NULL) + { + *string = ""; + } + else if (string < (char**)msg || + string > (char**)msg + sizeof(stroke_msg_t) || + *string < (char*)msg->buffer - (u_int)msg || + *string > (char*)(u_int)msg->length) + { + *string = "(invalid char* in stroke msg)"; + } + else + { + *string = (char*)msg + (u_int)*string; + } +} + +/** + * Add a connection to the configuration list + */ +static void stroke_add_conn(private_stroke_t *this, stroke_msg_t *msg) +{ + connection_t *connection; + policy_t *policy; + identification_t *my_id, *other_id; + host_t *my_host, *other_host, *my_subnet, *other_subnet; + proposal_t *proposal; + traffic_selector_t *my_ts, *other_ts; + x509_t *cert; + + pop_string(msg, &msg->add_conn.name); + pop_string(msg, &msg->add_conn.me.address); + pop_string(msg, &msg->add_conn.other.address); + pop_string(msg, &msg->add_conn.me.id); + pop_string(msg, &msg->add_conn.other.id); + pop_string(msg, &msg->add_conn.me.cert); + pop_string(msg, &msg->add_conn.other.cert); + pop_string(msg, &msg->add_conn.me.subnet); + pop_string(msg, &msg->add_conn.other.subnet); + + this->logger->log(this->logger, CONTROL, "received stroke: add connection \"%s\"", msg->add_conn.name); + + my_host = host_create(AF_INET, msg->add_conn.me.address, 500); + if (my_host == NULL) + { + this->stroke_logger->log(this->stroke_logger, ERROR, "invalid host: %s", msg->add_conn.me.address); + return; + } + other_host = host_create(AF_INET, msg->add_conn.other.address, 500); + if (other_host == NULL) + { + this->stroke_logger->log(this->stroke_logger, ERROR, "invalid host: %s", msg->add_conn.other.address); + my_host->destroy(my_host); + return; + } + my_id = identification_create_from_string(*msg->add_conn.me.id ? + msg->add_conn.me.id : msg->add_conn.me.address); + if (my_id == NULL) + { + this->stroke_logger->log(this->stroke_logger, ERROR, "invalid id: %s", msg->add_conn.me.id); + my_host->destroy(my_host); + other_host->destroy(other_host); + return; + } + other_id = identification_create_from_string(*msg->add_conn.other.id ? + msg->add_conn.other.id : msg->add_conn.other.address); + if (other_id == NULL) + { + my_host->destroy(my_host); + other_host->destroy(other_host); + my_id->destroy(my_id); + this->stroke_logger->log(this->stroke_logger, ERROR, "invalid id: %s", msg->add_conn.other.id); + return; + } + + my_subnet = host_create(AF_INET, *msg->add_conn.me.subnet ? msg->add_conn.me.subnet : msg->add_conn.me.address, 500); + if (my_subnet == NULL) + { + my_host->destroy(my_host); + other_host->destroy(other_host); + my_id->destroy(my_id); + other_id->destroy(other_id); + this->stroke_logger->log(this->stroke_logger, ERROR, "invalid subnet: %s", msg->add_conn.me.subnet); + return; + } + + other_subnet = host_create(AF_INET, *msg->add_conn.other.subnet ? msg->add_conn.other.subnet : msg->add_conn.other.address, 500); + if (other_subnet == NULL) + { + my_host->destroy(my_host); + other_host->destroy(other_host); + my_id->destroy(my_id); + other_id->destroy(other_id); + my_subnet->destroy(my_subnet); + this->stroke_logger->log(this->stroke_logger, ERROR, "invalid subnet: %s", msg->add_conn.me.subnet); + return; + } + + my_ts = traffic_selector_create_from_subnet(my_subnet, *msg->add_conn.me.subnet ? msg->add_conn.me.subnet_mask : 32); + my_subnet->destroy(my_subnet); + other_ts = traffic_selector_create_from_subnet(other_subnet, *msg->add_conn.other.subnet ? msg->add_conn.other.subnet_mask : 32); + other_subnet->destroy(other_subnet); + + if (charon->socket->is_listening_on(charon->socket, other_host)) + { + this->stroke_logger->log(this->stroke_logger, CONTROL|LEVEL1, "left is other host, switching"); + + host_t *tmp_host = my_host; + identification_t *tmp_id = my_id; + traffic_selector_t *tmp_ts = my_ts; + char *tmp_cert = msg->add_conn.me.cert; + + my_host = other_host; + other_host = tmp_host; + my_id = other_id; + other_id = tmp_id; + my_ts = other_ts; + other_ts = tmp_ts; + msg->add_conn.me.cert = msg->add_conn.other.cert; + msg->add_conn.other.cert = tmp_cert; + } + else if (charon->socket->is_listening_on(charon->socket, my_host)) + { + this->stroke_logger->log(this->stroke_logger, CONTROL|LEVEL1, "left is own host, not switching"); + } + else + { + this->stroke_logger->log(this->stroke_logger, ERROR, "left nor right host is our, aborting"); + + my_host->destroy(my_host); + other_host->destroy(other_host); + my_id->destroy(my_id); + other_id->destroy(other_id); + my_ts->destroy(my_ts); + other_ts->destroy(other_ts); + return; + } + + if (msg->add_conn.me.cert) + { + char file[128]; + snprintf(file, sizeof(file), "%s/%s", CERTIFICATE_DIR, msg->add_conn.me.cert); + cert = x509_create_from_file(file); + if (cert) + { + my_id->destroy(my_id); + my_id = cert->get_subject(cert); + my_id = my_id->clone(my_id); + cert->destroy(cert); + this->logger->log(this->logger, CONTROL, + "valid certificate with ID \"%s\"", + my_id->get_string(my_id)); + } + } + if (msg->add_conn.other.cert) + { + char file[128]; + snprintf(file, sizeof(file), "%s/%s", CERTIFICATE_DIR, msg->add_conn.other.cert); + cert = x509_create_from_file(file); + if (cert) + { + other_id->destroy(other_id); + other_id = cert->get_subject(cert); + other_id = other_id->clone(other_id); + cert->destroy(cert); + this->logger->log(this->logger, CONTROL, + "valid certificate with ID \"%s\"", + other_id->get_string(other_id)); + } + } + + connection = connection_create(msg->add_conn.name, + my_host, other_host, + my_id->clone(my_id), other_id->clone(other_id), + RSA_DIGITAL_SIGNATURE); + proposal = proposal_create(1); + proposal->add_algorithm(proposal, PROTO_IKE, ENCRYPTION_ALGORITHM, ENCR_AES_CBC, 16); + proposal->add_algorithm(proposal, PROTO_IKE, INTEGRITY_ALGORITHM, AUTH_HMAC_SHA1_96, 0); + proposal->add_algorithm(proposal, PROTO_IKE, INTEGRITY_ALGORITHM, AUTH_HMAC_MD5_96, 0); + proposal->add_algorithm(proposal, PROTO_IKE, PSEUDO_RANDOM_FUNCTION, PRF_HMAC_SHA1, 0); + proposal->add_algorithm(proposal, PROTO_IKE, PSEUDO_RANDOM_FUNCTION, PRF_HMAC_MD5, 0); + proposal->add_algorithm(proposal, PROTO_IKE, DIFFIE_HELLMAN_GROUP, MODP_2048_BIT, 0); + proposal->add_algorithm(proposal, PROTO_IKE, DIFFIE_HELLMAN_GROUP, MODP_1536_BIT, 0); + proposal->add_algorithm(proposal, PROTO_IKE, DIFFIE_HELLMAN_GROUP, MODP_1024_BIT, 0); + proposal->add_algorithm(proposal, PROTO_IKE, DIFFIE_HELLMAN_GROUP, MODP_4096_BIT, 0); + proposal->add_algorithm(proposal, PROTO_IKE, DIFFIE_HELLMAN_GROUP, MODP_8192_BIT, 0); + connection->add_proposal(connection, proposal); + /* add to global connection list */ + charon->connections->add_connection(charon->connections, connection); + + policy = policy_create(my_id, other_id); + proposal = proposal_create(1); + proposal->add_algorithm(proposal, PROTO_ESP, ENCRYPTION_ALGORITHM, ENCR_AES_CBC, 16); + proposal->add_algorithm(proposal, PROTO_ESP, INTEGRITY_ALGORITHM, AUTH_HMAC_SHA1_96, 0); + proposal->add_algorithm(proposal, PROTO_ESP, INTEGRITY_ALGORITHM, AUTH_HMAC_MD5_96, 0); + policy->add_proposal(policy, proposal); + policy->add_my_traffic_selector(policy, my_ts); + policy->add_other_traffic_selector(policy, other_ts); + /* add to global policy list */ + charon->policies->add_policy(charon->policies, policy); + + this->stroke_logger->log(this->stroke_logger, CONTROL|LEVEL1, "connection \"%s\" added", msg->add_conn.name); +} + +/** + * initiate a connection by name + */ +static void stroke_initiate(private_stroke_t *this, stroke_msg_t *msg) +{ + initiate_ike_sa_job_t *job; + connection_t *connection; + + pop_string(msg, &(msg->initiate.name)); + this->logger->log(this->logger, CONTROL, "received stroke: initiate \"%s\"", msg->initiate.name); + connection = charon->connections->get_connection_by_name(charon->connections, msg->initiate.name); + if (connection == NULL) + { + this->stroke_logger->log(this->stroke_logger, ERROR, "no connection named \"%s\"", msg->initiate.name); + } + else + { + job = initiate_ike_sa_job_create(connection); + charon->job_queue->add(charon->job_queue, (job_t*)job); + } +} + +/** + * terminate a connection by name + */ +static void stroke_terminate(private_stroke_t *this, stroke_msg_t *msg) +{ + linked_list_t *ike_sas; + iterator_t *iterator; + int instances = 0; + + pop_string(msg, &(msg->terminate.name)); + this->logger->log(this->logger, CONTROL, "received stroke: terminate \"%s\"", msg->terminate.name); + + ike_sas = charon->ike_sa_manager->get_ike_sa_list_by_name(charon->ike_sa_manager, msg->terminate.name); + + iterator = ike_sas->create_iterator(ike_sas, TRUE); + while (iterator->has_next(iterator)) + { + ike_sa_id_t *ike_sa_id; + iterator->current(iterator, (void**)&ike_sa_id); + charon->ike_sa_manager->delete(charon->ike_sa_manager, ike_sa_id); + ike_sa_id->destroy(ike_sa_id); + instances++; + } + iterator->destroy(iterator); + ike_sas->destroy(ike_sas); + this->stroke_logger->log(this->stroke_logger, CONTROL, "terminated %d instances of %s", instances, msg->terminate.name); +} + +/** + * show status of (established) connections + */ +static void stroke_status(private_stroke_t *this, stroke_msg_t *msg) +{ + if (msg->status.name) + { + pop_string(msg, &(msg->status.name)); + } + charon->ike_sa_manager->log_status(charon->ike_sa_manager, this->stroke_logger, msg->status.name); +} + +logger_context_t get_context(char *context) +{ + if (strcasecmp(context, "ALL") == 0) return ALL_LOGGERS; + else if (strcasecmp(context, "PARSR") == 0) return PARSER; + else if (strcasecmp(context, "GNRAT") == 0) return GENERATOR; + else if (strcasecmp(context, "IKESA") == 0) return IKE_SA; + else if (strcasecmp(context, "SAMGR") == 0) return IKE_SA_MANAGER; + else if (strcasecmp(context, "CHDSA") == 0) return CHILD_SA; + else if (strcasecmp(context, "MESSG") == 0) return MESSAGE; + else if (strcasecmp(context, "TPOOL") == 0) return THREAD_POOL; + else if (strcasecmp(context, "WORKR") == 0) return WORKER; + else if (strcasecmp(context, "SCHED") == 0) return SCHEDULER; + else if (strcasecmp(context, "SENDR") == 0) return SENDER; + else if (strcasecmp(context, "RECVR") == 0) return RECEIVER; + else if (strcasecmp(context, "SOCKT") == 0) return SOCKET; + else if (strcasecmp(context, "TESTR") == 0) return TESTER; + else if (strcasecmp(context, "DAEMN") == 0) return DAEMON; + else if (strcasecmp(context, "CONFG") == 0) return CONFIG; + else if (strcasecmp(context, "ENCPL") == 0) return ENCRYPTION_PAYLOAD; + else if (strcasecmp(context, "PAYLD") == 0) return PAYLOAD; + else return -2; +} + +/** + * set the type of logged messages in a context + */ +static void stroke_logtype(private_stroke_t *this, stroke_msg_t *msg) +{ + pop_string(msg, &(msg->logtype.context)); + pop_string(msg, &(msg->logtype.type)); + + this->logger->log(this->logger, CONTROL, "received stroke: logtype for %s", msg->logtype.context); + + log_level_t level; + logger_context_t context = get_context(msg->logtype.context); + if (context == -2) + { + this->stroke_logger->log(this->stroke_logger, ERROR, "invalid context (%s)!", msg->logtype.context); + return; + } + + if (strcasecmp(msg->logtype.type, "CONTROL") == 0) level = CONTROL; + else if (strcasecmp(msg->logtype.type, "ERROR") == 0) level = ERROR; + else if (strcasecmp(msg->logtype.type, "AUDIT") == 0) level = AUDIT; + else if (strcasecmp(msg->logtype.type, "RAW") == 0) level = RAW; + else if (strcasecmp(msg->logtype.type, "PRIVATE") == 0) level = PRIVATE; + else + { + this->stroke_logger->log(this->stroke_logger, ERROR, "invalid type (%s)!", msg->logtype.type); + return; + } + + if (msg->logtype.enable) + { + logger_manager->enable_log_level(logger_manager, context, level); + } + else + { + logger_manager->disable_log_level(logger_manager, context, level); + } +} + +/** + * set the verbosity of a logger + */ +static void stroke_loglevel(private_stroke_t *this, stroke_msg_t *msg) +{ + pop_string(msg, &(msg->loglevel.context)); + + this->logger->log(this->logger, CONTROL, "received stroke: loglevel for %s", msg->loglevel.context); + + log_level_t level; + logger_context_t context = get_context(msg->loglevel.context); + + if (context == -2) + { + this->stroke_logger->log(this->stroke_logger, ERROR, "invalid context (%s)!", msg->loglevel.context); + return; + } + + if (msg->loglevel.level == 0) + { + level = LEVEL0; + } + else if (msg->loglevel.level == 1) + { + level = LEVEL1; + } + else if (msg->loglevel.level == 2) + { + level = LEVEL2; + } + else if (msg->loglevel.level == 3) + { + level = LEVEL3; + } + else + { + this->stroke_logger->log(this->stroke_logger, ERROR, "invalid level (%d)!", msg->loglevel.level); + return; + } + + logger_manager->enable_log_level(logger_manager, context, level); +} + +/** + * Implementation of private_stroke_t.stroke_receive. + */ +static void stroke_receive(private_stroke_t *this) +{ + stroke_msg_t *msg; + u_int16_t msg_length; + struct sockaddr_un strokeaddr; + int strokeaddrlen = sizeof(strokeaddr); + ssize_t bytes_read; + int strokefd; + FILE *strokefile; + int oldstate; + + /* disable cancellation by default */ + pthread_setcancelstate(PTHREAD_CANCEL_DISABLE, NULL); + + while (1) + { + /* wait for connections, but allow thread to terminate */ + pthread_setcancelstate(PTHREAD_CANCEL_ENABLE, &oldstate); + strokefd = accept(this->socket, (struct sockaddr *)&strokeaddr, &strokeaddrlen); + pthread_setcancelstate(oldstate, NULL); + + if (strokefd < 0) + { + this->logger->log(this->logger, ERROR, "accepting stroke connection failed: %s", strerror(errno)); + continue; + } + + /* peek the length */ + bytes_read = recv(strokefd, &msg_length, sizeof(msg_length), MSG_PEEK); + if (bytes_read != sizeof(msg_length)) + { + this->logger->log(this->logger, ERROR, "reading lenght of stroke message failed"); + close(strokefd); + continue; + } + + /* read message */ + msg = malloc(msg_length); + bytes_read = recv(strokefd, msg, msg_length, 0); + if (bytes_read != msg_length) + { + this->logger->log(this->logger, ERROR, "reading stroke message failed: %s"); + close(strokefd); + continue; + } + + strokefile = fdopen(dup(strokefd), "w"); + if (strokefile == NULL) + { + this->logger->log(this->logger, ERROR, "opening stroke output channel failed:", strerror(errno)); + close(strokefd); + free(msg); + continue; + } + + /* setup a logger which writes status to the unix socket */ + this->stroke_logger = logger_create("-", CONTROL|ERROR, FALSE, strokefile); + + this->logger->log_bytes(this->logger, RAW, "stroke message", (void*)msg, msg_length); + + switch (msg->type) + { + case STR_INITIATE: + { + stroke_initiate(this, msg); + break; + } + case STR_TERMINATE: + { + stroke_terminate(this, msg); + break; + } + case STR_STATUS: + { + stroke_status(this, msg); + break; + } + case STR_STATUS_ALL: + { + this->stroke_logger->enable_level(this->stroke_logger, LEVEL1); + stroke_status(this, msg); + break; + } + case STR_ADD_CONN: + { + stroke_add_conn(this, msg); + break; + } + case STR_LOGTYPE: + { + stroke_logtype(this, msg); + break; + } + case STR_LOGLEVEL: + { + stroke_loglevel(this, msg); + break; + } + default: + this->logger->log(this->logger, ERROR, "received invalid stroke"); + } + this->stroke_logger->destroy(this->stroke_logger); + fclose(strokefile); + close(strokefd); + free(msg); + } +} + +/** + * Implementation of stroke_t.destroy. + */ +static void destroy(private_stroke_t *this) +{ + + pthread_cancel(this->assigned_thread); + pthread_join(this->assigned_thread, NULL); + + close(this->socket); + unlink(socket_addr.sun_path); + free(this); +} + + +/* + * Described in header-file + */ +stroke_t *stroke_create() +{ + private_stroke_t *this = malloc_thing(private_stroke_t); + mode_t old; + + /* public functions */ + this->public.destroy = (void (*)(stroke_t*))destroy; + + /* private functions */ + this->stroke_receive = stroke_receive; + + this->logger = logger_manager->get_logger(logger_manager, CONFIG); + + /* set up unix socket */ + this->socket = socket(AF_UNIX, SOCK_STREAM, 0); + if (this->socket == -1) + { + this->logger->log(this->logger, ERROR, "could not create whack socket"); + free(this); + return NULL; + } + + old = umask(~S_IRWXU); + if (bind(this->socket, (struct sockaddr *)&socket_addr, sizeof(socket_addr)) < 0) + { + this->logger->log(this->logger, ERROR, "could not bind stroke socket: %s", strerror(errno)); + close(this->socket); + free(this); + return NULL; + } + umask(old); + + if (listen(this->socket, 0) < 0) + { + this->logger->log(this->logger, ERROR, "could not listen on stroke socket: %s", strerror(errno)); + close(this->socket); + unlink(socket_addr.sun_path); + free(this); + return NULL; + } + + /* start a thread reading from the socket */ + if (pthread_create(&(this->assigned_thread), NULL, (void*(*)(void*))this->stroke_receive, this) != 0) + { + this->logger->log(this->logger, ERROR, "Could not spawn stroke thread"); + close(this->socket); + unlink(socket_addr.sun_path); + free(this); + return NULL; + } + + return (&this->public); +} diff --git a/src/charon/threads/stroke_interface.h b/src/charon/threads/stroke_interface.h new file mode 100644 index 000000000..16eb66533 --- /dev/null +++ b/src/charon/threads/stroke_interface.h @@ -0,0 +1,76 @@ +/** + * @file stroke.h + * + * @brief Interface of stroke_t. + * + */ + +/* + * Copyright (C) 2006 Martin Willi + * Hochschule fuer Technik Rapperswil + * + * This program is free software; you can redistribute it and/or modify it + * under the terms of the GNU General Public License as published by the + * Free Software Foundation; either version 2 of the License, or (at your + * option) any later version. See <http://www.fsf.org/copyleft/gpl.txt>. + * + * 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 General Public License + * for more details. + */ + +#ifndef STROKE_INTERFACE_H_ +#define STROKE_INTERFACE_H_ + +#include <config/policies/policy_store.h> +#include <config/connections/connection_store.h> +#include <config/credentials/credential_store.h> + + +typedef struct stroke_t stroke_t; + +/** + * @brief Stroke is a configuration and control interface which + * allows other processes to modify charons behavior. + * + * stroke_t allows config manipulation (as whack in pluto). + * Messages of type stroke_msg_t's are sent over a unix socket + * (/var/run/charon.ctl). stroke_t implements the connections_t + * and the policies_t interface, which means it acts as a + * configuration backend for those too. stroke_t uses an own + * thread to read from the socket. + * + * @warning DO NOT cast stroke_t to any of the implemented interfaces! + * stroke_t implements multiple interfaces, so you must use + * stroke_t.interface_xy to access the specific interface! You have + * been warned... + * + * @todo Add clean thread cancellation + * + * @b Constructors: + * - stroke_create() + * + * @ingroup threads + */ +struct stroke_t { + + /** + * @brief Destroy a stroke_t instance. + * + * @param this stroke_t objec to destroy + */ + void (*destroy) (stroke_t *this); +}; + + +/** + * @brief Create the stroke interface and listen on the socket. + * + * @return stroke_t object + * + * @ingroup threads + */ +stroke_t *stroke_create(void); + +#endif /* STROKE_INTERFACE_H_ */ diff --git a/src/charon/threads/thread_pool.c b/src/charon/threads/thread_pool.c new file mode 100644 index 000000000..4482e795f --- /dev/null +++ b/src/charon/threads/thread_pool.c @@ -0,0 +1,623 @@ +/** + * @file thread_pool.c + * + * @brief Implementation of thread_pool_t. + * + */ + +/* + * Copyright (C) 2005 Jan Hutter, Martin Willi + * Hochschule fuer Technik Rapperswil + * + * This program is free software; you can redistribute it and/or modify it + * under the terms of the GNU General Public License as published by the + * Free Software Foundation; either version 2 of the License, or (at your + * option) any later version. See <http://www.fsf.org/copyleft/gpl.txt>. + * + * 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 General Public License + * for more details. + */ + +#include <stdlib.h> +#include <pthread.h> +#include <string.h> +#include <errno.h> + +#include "thread_pool.h" + +#include <daemon.h> +#include <queues/job_queue.h> +#include <queues/jobs/delete_half_open_ike_sa_job.h> +#include <queues/jobs/delete_established_ike_sa_job.h> +#include <queues/jobs/incoming_packet_job.h> +#include <queues/jobs/initiate_ike_sa_job.h> +#include <queues/jobs/retransmit_request_job.h> +#include <encoding/payloads/notify_payload.h> +#include <utils/logger.h> + + +typedef struct private_thread_pool_t private_thread_pool_t; + +/** + * @brief Private data of thread_pool_t class. + */ +struct private_thread_pool_t { + /** + * Public thread_pool_t interface. + */ + thread_pool_t public; + + /** + * @brief Main processing function for worker threads. + * + * Gets a job from the job queue and calls corresponding + * function for processing. + * + * @param this calling object + */ + void (*process_jobs) (private_thread_pool_t *this); + + /** + * @brief Process a INCOMING_PACKET job. + * + * @param this calling object + * @param job incoming_packet_job_t object + */ + void (*process_incoming_packet_job) (private_thread_pool_t *this, incoming_packet_job_t *job); + + /** + * @brief Process a INITIATE_IKE_SA job. + * + * @param this calling object + * @param job initiate_ike_sa_job_t object + */ + void (*process_initiate_ike_sa_job) (private_thread_pool_t *this, initiate_ike_sa_job_t *job); + + /** + * @brief Process a DELETE_HALF_OPEN_IKE_SA job. + * + * @param this calling object + * @param job delete__half_open_ike_sa_job_t object + */ + void (*process_delete_half_open_ike_sa_job) (private_thread_pool_t *this, delete_half_open_ike_sa_job_t *job); + + /** + * @brief Process a DELETE_ESTABLISHED_IKE_SA job. + * + * @param this calling object + * @param job delete_established_ike_sa_job_t object + */ + void (*process_delete_established_ike_sa_job) (private_thread_pool_t *this, delete_established_ike_sa_job_t *job); + + /** + * @brief Process a RETRANSMIT_REQUEST job. + * + * @param this calling object + * @param job retransmit_request_job_t object + */ + void (*process_retransmit_request_job) (private_thread_pool_t *this, retransmit_request_job_t *job); + + /** + * Creates a job of type DELETE_HALF_OPEN_IKE_SA. + * + * This job is used to delete IKE_SA's which are still in state INITIATOR_INIT, + * RESPONDER_INIT, IKE_AUTH_REQUESTED, IKE_INIT_REQUESTED or IKE_INIT_RESPONDED. + * + * @param ike_sa_id ID of IKE_SA to delete + * @param delay Delay in ms after a half open IKE_SA gets deleted! + */ + void (*create_delete_half_open_ike_sa_job) (private_thread_pool_t *this,ike_sa_id_t *ike_sa_id, u_int32_t delay); + + /** + * Number of running threads. + */ + size_t pool_size; + + /** + * Array of thread ids. + */ + pthread_t *threads; + + /** + * Logger of the thread pool. + */ + logger_t *pool_logger; + + /** + * Logger of the worker threads. + */ + logger_t *worker_logger; +} ; + +/** + * Implementation of private_thread_pool_t.process_jobs. + */ +static void process_jobs(private_thread_pool_t *this) +{ + job_t *job; + job_type_t job_type; + timeval_t start_time; + timeval_t end_time; + + /* cancellation disabled by default */ + pthread_setcancelstate(PTHREAD_CANCEL_DISABLE, NULL); + + this->worker_logger->log(this->worker_logger, CONTROL, "Worker thread running, thread_id: %u", (int)pthread_self()); + + for (;;) { + + job = charon->job_queue->get(charon->job_queue); + job_type = job->get_type(job); + this->worker_logger->log(this->worker_logger, CONTROL|LEVEL2, "Process job of type %s", + mapping_find(job_type_m,job_type)); + gettimeofday(&start_time,NULL); + switch (job_type) + { + case INCOMING_PACKET: + { + this->process_incoming_packet_job(this, (incoming_packet_job_t*)job); + job->destroy(job); + break; + } + case INITIATE_IKE_SA: + { + this->process_initiate_ike_sa_job(this, (initiate_ike_sa_job_t*)job); + job->destroy(job); + break; + } + case DELETE_HALF_OPEN_IKE_SA: + { + this->process_delete_half_open_ike_sa_job(this, (delete_half_open_ike_sa_job_t*)job); + job->destroy(job); + break; + } + case DELETE_ESTABLISHED_IKE_SA: + { + this->process_delete_established_ike_sa_job(this, (delete_established_ike_sa_job_t*)job); + job->destroy(job); + break; + } + case RETRANSMIT_REQUEST: + { + this->process_retransmit_request_job(this, (retransmit_request_job_t*)job); + break; + } + default: + { + this->worker_logger->log(this->worker_logger, ERROR, "Job of type %s not supported!", + mapping_find(job_type_m,job_type)); + job->destroy(job); + break; + } + } + gettimeofday(&end_time,NULL); + + this->worker_logger->log(this->worker_logger, CONTROL | LEVEL2, "Processed job of type %s in %d us", + mapping_find(job_type_m,job_type), + (((end_time.tv_sec - start_time.tv_sec) * 1000000) + (end_time.tv_usec - start_time.tv_usec))); + + + } +} + +/** + * Implementation of private_thread_pool_t.process_incoming_packet_job. + */ +static void process_incoming_packet_job(private_thread_pool_t *this, incoming_packet_job_t *job) +{ + packet_t *packet; + message_t *message; + ike_sa_t *ike_sa; + ike_sa_id_t *ike_sa_id; + status_t status; + + + packet = job->get_packet(job); + + message = message_create_from_packet(packet); + + status = message->parse_header(message); + if (status != SUCCESS) + { + this->worker_logger->log(this->worker_logger, ERROR, "Message header could not be verified!"); + message->destroy(message); + return; + } + + this->worker_logger->log(this->worker_logger, CONTROL|LEVEL2, "Message is a %s %s", + mapping_find(exchange_type_m, message->get_exchange_type(message)), + message->get_request(message) ? "request" : "reply"); + + if ((message->get_major_version(message) != IKE_MAJOR_VERSION) || + (message->get_minor_version(message) != IKE_MINOR_VERSION)) + { + this->worker_logger->log(this->worker_logger, ERROR | LEVEL2, "IKE version %d.%d not supported", + message->get_major_version(message), + message->get_minor_version(message)); + /* + * This check is not handled in state_t object of IKE_SA to increase speed. + */ + if ((message->get_exchange_type(message) == IKE_SA_INIT) && (message->get_request(message))) + { + message_t *response; + message->get_ike_sa_id(message, &ike_sa_id); + ike_sa_id->switch_initiator(ike_sa_id); + response = message_create_notify_reply(message->get_destination(message), + message->get_source(message), + IKE_SA_INIT, + FALSE,ike_sa_id,INVALID_MAJOR_VERSION); + message->destroy(message); + ike_sa_id->destroy(ike_sa_id); + status = response->generate(response, NULL, NULL, &packet); + if (status != SUCCESS) + { + this->worker_logger->log(this->worker_logger, ERROR, "Could not generate packet from message"); + response->destroy(response); + return; + } + this->worker_logger->log(this->worker_logger, ERROR, "Send notify reply of type INVALID_MAJOR_VERSION"); + charon->send_queue->add(charon->send_queue, packet); + response->destroy(response); + return; + } + message->destroy(message); + return; + } + + message->get_ike_sa_id(message, &ike_sa_id); + + ike_sa_id->switch_initiator(ike_sa_id); + + this->worker_logger->log(this->worker_logger, CONTROL|LEVEL3, "Checking out IKE SA %lld:%lld, role %s", + ike_sa_id->get_initiator_spi(ike_sa_id), + ike_sa_id->get_responder_spi(ike_sa_id), + ike_sa_id->is_initiator(ike_sa_id) ? "initiator" : "responder"); + + status = charon->ike_sa_manager->checkout(charon->ike_sa_manager,ike_sa_id, &ike_sa); + if ((status != SUCCESS) && (status != CREATED)) + { + this->worker_logger->log(this->worker_logger, ERROR, "IKE SA could not be checked out"); + ike_sa_id->destroy(ike_sa_id); + message->destroy(message); + + /* + * TODO send notify reply of type INVALID_IKE_SPI if SPI could not be found ? + */ + + return; + } + + if (status == CREATED) + { + this->worker_logger->log(this->worker_logger, CONTROL|LEVEL3, + "Create Job to delete half open IKE_SA."); + this->create_delete_half_open_ike_sa_job(this,ike_sa_id, + charon->configuration->get_half_open_ike_sa_timeout(charon->configuration)); + } + + status = ike_sa->process_message(ike_sa, message); + + this->worker_logger->log(this->worker_logger, CONTROL|LEVEL3, "%s IKE SA %lld:%lld, role %s", + (status == DELETE_ME) ? "Checkin and delete" : "Checkin", + ike_sa_id->get_initiator_spi(ike_sa_id), + ike_sa_id->get_responder_spi(ike_sa_id), + ike_sa_id->is_initiator(ike_sa_id) ? "initiator" : "responder"); + ike_sa_id->destroy(ike_sa_id); + + if (status == DELETE_ME) + { + status = charon->ike_sa_manager->checkin_and_delete(charon->ike_sa_manager, ike_sa); + } + else + { + status = charon->ike_sa_manager->checkin(charon->ike_sa_manager, ike_sa); + } + + if (status != SUCCESS) + { + this->worker_logger->log(this->worker_logger, ERROR, "Checkin of IKE SA failed!"); + } + message->destroy(message); +} + +/** + * Implementation of private_thread_pool_t.process_initiate_ike_sa_job. + */ +static void process_initiate_ike_sa_job(private_thread_pool_t *this, initiate_ike_sa_job_t *job) +{ + /* + * Initiatie an IKE_SA: + * - is defined by a name of a configuration + * - create an empty IKE_SA via manager + * - call initiate_connection on this sa + */ + ike_sa_t *ike_sa; + status_t status; + + + this->worker_logger->log(this->worker_logger, CONTROL|LEVEL2, "Creating and checking out IKE SA"); + charon->ike_sa_manager->create_and_checkout(charon->ike_sa_manager, &ike_sa); + + status = ike_sa->initiate_connection(ike_sa, job->get_connection(job)); + if (status != SUCCESS) + { + this->worker_logger->log(this->worker_logger, ERROR, "Initiation returned %s, going to delete IKE_SA.", + mapping_find(status_m, status)); + charon->ike_sa_manager->checkin_and_delete(charon->ike_sa_manager, ike_sa); + return; + } + + this->worker_logger->log(this->worker_logger, CONTROL|LEVEL3, "Create Job to delete half open IKE_SA."); + this->create_delete_half_open_ike_sa_job(this,ike_sa->get_id(ike_sa), + charon->configuration->get_half_open_ike_sa_timeout(charon->configuration)); + + this->worker_logger->log(this->worker_logger, CONTROL|LEVEL2, "Checking in IKE SA"); + status = charon->ike_sa_manager->checkin(charon->ike_sa_manager, ike_sa); + if (status != SUCCESS) + { + this->worker_logger->log(this->worker_logger, ERROR, "Could not checkin IKE_SA (%s)", + mapping_find(status_m, status)); + } +} + +/** + * Implementation of private_thread_pool_t.process_delete_ike_sa_job. + */ +static void process_delete_half_open_ike_sa_job(private_thread_pool_t *this, delete_half_open_ike_sa_job_t *job) +{ + ike_sa_id_t *ike_sa_id = job->get_ike_sa_id(job); + ike_sa_t *ike_sa; + status_t status; + status = charon->ike_sa_manager->checkout(charon->ike_sa_manager,ike_sa_id, &ike_sa); + if ((status != SUCCESS) && (status != CREATED)) + { + this->worker_logger->log(this->worker_logger, CONTROL | LEVEL3, "IKE SA seems to be allready deleted and so doesn't have to be deleted"); + return; + } + + + switch (ike_sa->get_state(ike_sa)) + { + case INITIATOR_INIT: + case RESPONDER_INIT: + case IKE_SA_INIT_REQUESTED: + case IKE_SA_INIT_RESPONDED: + case IKE_AUTH_REQUESTED: + { + /* IKE_SA is half open and gets deleted! */ + status = charon->ike_sa_manager->checkin_and_delete(charon->ike_sa_manager, ike_sa); + if (status != SUCCESS) + { + this->worker_logger->log(this->worker_logger, ERROR, "Could not checkin and delete checked out IKE_SA!"); + } + break; + } + default: + { + /* IKE_SA is established and so is not getting deleted! */ + status = charon->ike_sa_manager->checkin(charon->ike_sa_manager, ike_sa); + if (status != SUCCESS) + { + this->worker_logger->log(this->worker_logger, ERROR, "Could not checkin a checked out IKE_SA!"); + } + break; + } + } +} + +/** + * Implementation of private_thread_pool_t.process_delete_established_ike_sa_job. + */ +static void process_delete_established_ike_sa_job(private_thread_pool_t *this, delete_established_ike_sa_job_t *job) +{ + ike_sa_id_t *ike_sa_id = job->get_ike_sa_id(job); + ike_sa_t *ike_sa; + status_t status; + status = charon->ike_sa_manager->checkout(charon->ike_sa_manager,ike_sa_id, &ike_sa); + if ((status != SUCCESS) && (status != CREATED)) + { + this->worker_logger->log(this->worker_logger, CONTROL | LEVEL3, "IKE SA seems to be allready deleted and so doesn't have to be deleted"); + return; + } + + switch (ike_sa->get_state(ike_sa)) + { + case INITIATOR_INIT: + case RESPONDER_INIT: + case IKE_SA_INIT_REQUESTED: + case IKE_SA_INIT_RESPONDED: + case IKE_AUTH_REQUESTED: + { + break; + } + default: + { + this->worker_logger->log(this->worker_logger, CONTROL, "Send delete request for IKE_SA."); + ike_sa->send_delete_ike_sa_request(ike_sa); + break; + } + } + this->worker_logger->log(this->worker_logger, CONTROL, "Delete established IKE_SA."); + status = charon->ike_sa_manager->checkin_and_delete(charon->ike_sa_manager, ike_sa); + if (status != SUCCESS) + { + this->worker_logger->log(this->worker_logger, ERROR, "Could not checkin and delete checked out IKE_SA!"); + } +} + + +/** + * Implementation of private_thread_pool_t.process_retransmit_request_job. + */ +static void process_retransmit_request_job(private_thread_pool_t *this, retransmit_request_job_t *job) +{ + + ike_sa_id_t *ike_sa_id = job->get_ike_sa_id(job); + u_int32_t message_id = job->get_message_id(job); + bool stop_retransmitting = FALSE; + u_int32_t timeout; + ike_sa_t *ike_sa; + status_t status; + + this->worker_logger->log(this->worker_logger, CONTROL|LEVEL2, "Checking out IKE SA %lld:%lld, role %s", + ike_sa_id->get_initiator_spi(ike_sa_id), + ike_sa_id->get_responder_spi(ike_sa_id), + ike_sa_id->is_initiator(ike_sa_id) ? "initiator" : "responder"); + + status = charon->ike_sa_manager->checkout(charon->ike_sa_manager,ike_sa_id, &ike_sa); + if ((status != SUCCESS) && (status != CREATED)) + { + job->destroy(job); + this->worker_logger->log(this->worker_logger, ERROR, "IKE SA could not be checked out. Allready deleted?"); + return; + } + + status = ike_sa->retransmit_request(ike_sa, message_id); + + if (status != SUCCESS) + { + this->worker_logger->log(this->worker_logger, CONTROL | LEVEL3, "Message doesn't have to be retransmitted"); + stop_retransmitting = TRUE; + } + + this->worker_logger->log(this->worker_logger, CONTROL|LEVEL2, "Checkin IKE SA %lld:%lld, role %s", + ike_sa_id->get_initiator_spi(ike_sa_id), + ike_sa_id->get_responder_spi(ike_sa_id), + ike_sa_id->is_initiator(ike_sa_id) ? "initiator" : "responder"); + + status = charon->ike_sa_manager->checkin(charon->ike_sa_manager, ike_sa); + if (status != SUCCESS) + { + this->worker_logger->log(this->worker_logger, ERROR, "Checkin of IKE SA failed!"); + } + + if (stop_retransmitting) + { + job->destroy(job); + return; + } + + job->increase_retransmit_count(job); + status = charon->configuration->get_retransmit_timeout (charon->configuration,job->get_retransmit_count(job),&timeout); + if (status != SUCCESS) + { + this->worker_logger->log(this->worker_logger, CONTROL | LEVEL2, "Message will not be anymore retransmitted"); + job->destroy(job); + /* + * TODO delete IKE_SA ? + */ + return; + } + charon->event_queue->add_relative(charon->event_queue,(job_t *) job,timeout); +} + + + +/** + * Implementation of private_thread_pool_t.create_delete_half_open_ike_sa_job. + */ +static void create_delete_half_open_ike_sa_job(private_thread_pool_t *this,ike_sa_id_t *ike_sa_id, u_int32_t delay) +{ + job_t *delete_job; + + this->worker_logger->log(this->worker_logger, CONTROL | LEVEL2, "Going to create job to delete half open IKE_SA in %d ms", delay); + + delete_job = (job_t *) delete_half_open_ike_sa_job_create(ike_sa_id); + charon->event_queue->add_relative(charon->event_queue,delete_job, delay); +} + + +/** + * Implementation of thread_pool_t.get_pool_size. + */ +static size_t get_pool_size(private_thread_pool_t *this) +{ + return this->pool_size; +} + +/** + * Implementation of thread_pool_t.destroy. + */ +static void destroy(private_thread_pool_t *this) +{ + int current; + /* flag thread for termination */ + for (current = 0; current < this->pool_size; current++) { + this->pool_logger->log(this->pool_logger, CONTROL, "cancelling worker thread #%d", current+1); + pthread_cancel(this->threads[current]); + } + + /* wait for all threads */ + for (current = 0; current < this->pool_size; current++) { + if (pthread_join(this->threads[current], NULL) == 0) + { + this->pool_logger->log(this->pool_logger, CONTROL, "worker thread #%d terminated", current+1); + } + else + { + this->pool_logger->log(this->pool_logger, ERROR, "could not terminate worker thread #%d", current+1); + } + } + + /* free mem */ + free(this->threads); + free(this); +} + +/* + * Described in header. + */ +thread_pool_t *thread_pool_create(size_t pool_size) +{ + int current; + + private_thread_pool_t *this = malloc_thing(private_thread_pool_t); + + /* fill in public fields */ + this->public.destroy = (void(*)(thread_pool_t*))destroy; + this->public.get_pool_size = (size_t(*)(thread_pool_t*))get_pool_size; + + this->process_jobs = process_jobs; + this->process_initiate_ike_sa_job = process_initiate_ike_sa_job; + this->process_delete_half_open_ike_sa_job = process_delete_half_open_ike_sa_job; + this->process_delete_established_ike_sa_job = process_delete_established_ike_sa_job; + this->process_incoming_packet_job = process_incoming_packet_job; + this->process_retransmit_request_job = process_retransmit_request_job; + this->create_delete_half_open_ike_sa_job = create_delete_half_open_ike_sa_job; + + this->pool_size = pool_size; + + this->threads = malloc(sizeof(pthread_t) * pool_size); + + this->pool_logger = logger_manager->get_logger(logger_manager, THREAD_POOL); + + this->worker_logger = logger_manager->get_logger(logger_manager, WORKER); + + /* try to create as many threads as possible, up tu pool_size */ + for (current = 0; current < pool_size; current++) + { + if (pthread_create(&(this->threads[current]), NULL, (void*(*)(void*))this->process_jobs, this) == 0) + { + this->pool_logger->log(this->pool_logger, CONTROL, "Created worker thread #%d", current+1); + } + else + { + /* creation failed, is it the first one? */ + if (current == 0) + { + this->pool_logger->log(this->pool_logger, ERROR, "Could not create any thread"); + free(this->threads); + free(this); + return NULL; + } + /* not all threads could be created, but at least one :-/ */ + this->pool_logger->log(this->pool_logger, ERROR, "Could only create %d from requested %d threads!", current, pool_size); + + this->pool_size = current; + return (thread_pool_t*)this; + } + } + return (thread_pool_t*)this; +} diff --git a/src/charon/threads/thread_pool.h b/src/charon/threads/thread_pool.h new file mode 100644 index 000000000..b33be08e3 --- /dev/null +++ b/src/charon/threads/thread_pool.h @@ -0,0 +1,78 @@ +/** + * @file thread_pool.h + * + * @brief Interface of thread_pool_t. + * + */ + +/* + * Copyright (C) 2005 Jan Hutter, Martin Willi + * Hochschule fuer Technik Rapperswil + * + * This program is free software; you can redistribute it and/or modify it + * under the terms of the GNU General Public License as published by the + * Free Software Foundation; either version 2 of the License, or (at your + * option) any later version. See <http://www.fsf.org/copyleft/gpl.txt>. + * + * 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 General Public License + * for more details. + */ + +#ifndef THREAD_POOL_H_ +#define THREAD_POOL_H_ + +#include <stdlib.h> + +#include <types.h> + + +typedef struct thread_pool_t thread_pool_t; + +/** + * @brief A thread_pool consists of a pool of threads processing jobs from the job queue. + * + * Current implementation uses as many threads as specified in constructor. + * A more improved version would dynamically increase thread count if necessary. + * + * @b Constructors: + * - thread_pool_create() + * + * @todo Add support for dynamic thread handling + * + * @ingroup threads + */ +struct thread_pool_t { + /** + * @brief Return currently instanciated thread count. + * + * @param thread_pool calling object + * @return size of thread pool + */ + size_t (*get_pool_size) (thread_pool_t *thread_pool); + + /** + * @brief Destroy a thread_pool_t object. + * + * Sends cancellation request to all threads and AWAITS their termination. + * + * @param thread_pool calling object + */ + void (*destroy) (thread_pool_t *thread_pool); +}; + +/** + * @brief Create the thread pool using using pool_size of threads. + * + * @param pool_size desired pool size + * @return + * - thread_pool_t object if one ore more threads could be started, or + * - NULL if no threads could be created + * + * @ingroup threads + */ +thread_pool_t *thread_pool_create(size_t pool_size); + + +#endif /*THREAD_POOL_H_*/ |