diff options
| -rw-r--r-- | ecp/server/dir.c | 15 | ||||
| -rw-r--r-- | ecp/server/dir.h | 1 | ||||
| -rw-r--r-- | ecp/server/server.c | 290 | ||||
| -rw-r--r-- | ecp/server/server.h | 12 | ||||
| -rw-r--r-- | ecp/src/ecp/dir/dir_client.c | 6 | ||||
| -rw-r--r-- | ecp/src/ecp/dir/dir_client.h | 2 |
6 files changed, 214 insertions, 112 deletions
diff --git a/ecp/server/dir.c b/ecp/server/dir.c index 95db4b5..c045599 100644 --- a/ecp/server/dir.c +++ b/ecp/server/dir.c @@ -428,7 +428,7 @@ void dir_process_item(ECPDirItem *dir_item, ECPSocket *sock, ecp_ecdh_public_t * pthread_mutex_lock(&node->mutex); pthread_rwlock_unlock(&dir_shadow_rwlock); - if (node->zombie) goto process_item_fin; + if (node->zombie || (s_public == NULL)) goto process_item_fin; if (!node->verified) { int key_ex = 0; @@ -754,6 +754,19 @@ void dit_init_serial(uint16_t serial) { } } +int dir_init_dir_cnt(unsigned int dir_cnt, uint16_t serial) { + int rv = ECP_OK; + + if (dir_online) { + pthread_rwlock_wrlock(&dir_online_rwlock); + if (dir_online->serial != serial) rv = ECP_ERR; + if (!rv) dir_vkey_req = dir_cnt / 2 + 1; + pthread_rwlock_unlock(&dir_online_rwlock); + } + + return rv; +} + void dir_announce(ECPSocket *sock, int ann_period) { struct hashtable_itr itr; DIRNode *node; diff --git a/ecp/server/dir.h b/ecp/server/dir.h index fee3a5e..dbcbc2e 100644 --- a/ecp/server/dir.h +++ b/ecp/server/dir.h @@ -69,6 +69,7 @@ void dir_remove_nodes(DIRNode *remove_node[], int remove_cnt); void dir_announce_allow(void); void dir_announce_block(void); void dit_init_serial(uint16_t serial); +int dir_init_dir_cnt(unsigned int dir_cnt, uint16_t serial); void dir_announce(ECPSocket *sock, int ann_period); int dir_start_announce(ECPSocket *sock); void dir_init_switch(ECPSocket *sock, int init_ann); diff --git a/ecp/server/server.c b/ecp/server/server.c index 3fc0e7c..a7140a1 100644 --- a/ecp/server/server.c +++ b/ecp/server/server.c @@ -10,6 +10,7 @@ #include <ecp/core.h> #include <ecp/vconn/vconn.h> #include <ecp/dir/dir.h> +#include <ecp/dir/dir_client.h> #include <util.h> @@ -23,8 +24,12 @@ static SRVConfig srv_config; static int proc_forked = 0; -static FILE *log_file = NULL; +static pthread_t rcvr_thd[MAX_THREADS]; +static pthread_mutex_t dir_sync_mutex; +static pthread_cond_t dir_sync_cond; +static int dir_sync_res = 0; +static FILE *log_file = NULL; int log_level; static const char *log_level_str[] = { "ERROR", @@ -40,6 +45,29 @@ static void handle_err(ECPConnection *conn, unsigned char mtype, int err) { LOG(LOG_ERR, "handle_err: ctype:%d mtype:%d err:%d\n", conn->type, mtype, err); } +static ssize_t conn_auth(ECPSocket *sock, ECPConnection *parent, unsigned char ctype, ecp_ecdh_public_t *public, unsigned char *msg, size_t msg_size) { + switch (ctype) { + case CTYPE_DIR: { + if (public == NULL) return ECP_ERR_AUTH; + if (!acl_inlist(public)) return ECP_ERR_AUTH; + return 0; + } + + case ECP_CTYPE_VLINK: { + if (public == NULL) return ECP_ERR_AUTH; + if ((parent == NULL) && !acl_inlist(public)) return ECP_ERR_AUTH; + return 0; + } + + case ECP_CTYPE_DIR: + case ECP_CTYPE_VCONN: + return 0; + + default: + return ECP_ERR_AUTH; + } +} + static ECPConnection *conn_new(ECPSocket *sock, ECPConnection *parent, unsigned char ctype) { ECPConnection *conn = NULL; @@ -93,35 +121,11 @@ static void conn_free(ECPConnection *conn) { free(conn); } -static ssize_t conn_auth(ECPSocket *sock, ECPConnection *parent, unsigned char ctype, ecp_ecdh_public_t *public, unsigned char *msg, size_t msg_size) { - switch (ctype) { - case CTYPE_DIR: { - if (public == NULL) return ECP_ERR_AUTH; - if (!acl_inlist(public)) return ECP_ERR_AUTH; - return 0; - } - - case ECP_CTYPE_VLINK: { - if (public == NULL) return ECP_ERR_AUTH; - if ((parent == NULL) && !acl_inlist(public)) return ECP_ERR_AUTH; - return 0; - } - - case ECP_CTYPE_DIR: - case ECP_CTYPE_VCONN: - return 0; - - default: - return ECP_ERR_AUTH; - } -} - static int core_logger(const char *format, ...) { va_list ap; - FILE *file = log_file ? log_file : stderr; va_start(ap, format); - log_vfprintf(LOG_ERR, file, format, ap); + log_vprintf(LOG_ERR, format, ap); va_end(ap); return 0; @@ -142,50 +146,56 @@ int ecp_init(ECPContext *ctx, ECPConnHandler *vconn_handler, ECPConnHandler *vli return ECP_OK; } -void log_vfprintf(int level, FILE *file, const char *format, va_list ap) { - time_t t; - char t_buf[26]; - char s_buf[256]; +void log_vprintf(int level, const char *format, va_list ap) { + FILE *file; if (level > log_level) return; if (level >= (sizeof(log_level_str) / sizeof(char *))) return; - t = time(NULL); - ctime_r(&t, t_buf); - /* terminate before newline */ - t_buf[24] = '\0'; - - vsnprintf(s_buf, sizeof(s_buf), format, ap); - - fprintf(file, "%s [%s]: %s", t_buf, log_level_str[level], s_buf); + file = srv_config.log_file ? srv_config.log_file : stderr; + if (proc_forked) { + time_t t; + char t_buf[26]; + char s_buf[256]; + int rv; + + t = time(NULL); + ctime_r(&t, t_buf); + /* terminate string before newline */ + t_buf[24] = '\0'; + + rv = vsnprintf(s_buf, sizeof(s_buf), format, ap); + /* always print newline at the and of string */ + if (rv <= 0) return; + if (rv >= (sizeof(s_buf) - 1)) { + s_buf[sizeof(s_buf) - 2] = '\n'; + } else if (s_buf[rv - 1] != '\n') { + s_buf[rv] = '\n'; + s_buf[rv + 1] = '\0'; + } + fprintf(file, "%s [%s]: %s", t_buf, log_level_str[level], s_buf); + } else { + vfprintf(file, format, ap); + } fflush(file); } void log_printf(int level, const char *format, ...) { va_list ap; - FILE *file = log_file ? log_file : stderr; if (level > log_level) return; va_start(ap, format); - log_vfprintf(level, file, format, ap); + log_vprintf(level, format, ap); va_end(ap); } static void fail(const char *format, ...) { va_list ap; - FILE *file; va_start(ap, format); - if (proc_forked) { - file = log_file ? log_file : stderr; - log_vfprintf(LOG_ERR, file, format, ap); - } else { - file = stderr; - vfprintf(file, format, ap); - } + log_vprintf(LOG_ERR, format, ap); va_end(ap); - fflush(file); exit(1); } @@ -196,6 +206,9 @@ static void usage(char *arg) { fprintf(stderr, "\t-d\n"); fprintf(stderr, "\t\tdetach\n"); + fprintf(stderr, "\t-t <number>\n"); + fprintf(stderr, "\t\tstart <number> receiver threads (maximum: %d)\n", MAX_THREADS); + fprintf(stderr, "\t-l <level>\n"); fprintf(stderr, "\t\tset log level: 0 - error, 1 - info, 2 - debug\n"); @@ -221,7 +234,7 @@ static void daemonize(void) { proc_forked = 1; - pid = setsid() + pid = setsid(); if (pid < 0) fail("setsid failed\n"); /* second fork */ @@ -229,14 +242,20 @@ static void daemonize(void) { if (pid < 0) fail("fork2 failed\n"); if (pid > 0) exit(0); - /* redirect stdio to /dev/null */ + /* redirect /dev/null to stdin */ null_rd = open("/dev/null", O_RDONLY); if (null_rd < 0) fail("open (r) /dev/null failed\n"); - null_wr = open("/dev/null", O_WRONLY); - if (null_wr < 0) fail("open (w) /dev/null failed\n"); - rv = dup2(null_rd, STDIN_FILENO); if (rv < 0) fail("dup2 (stdin) failed\n"); + close(null_rd); + + /* redirect stdout and stderr to srv_config.log_file or /dev/null */ + if (srv_config.log_file) { + null_wr = fileno(srv_config.log_file); + } else { + null_wr = open("/dev/null", O_WRONLY); + if (null_wr < 0) fail("open (w) /dev/null failed\n"); + } rv = dup2(null_wr, STDOUT_FILENO); if (rv < 0) fail("dup2 (stdout) failed\n"); @@ -244,44 +263,89 @@ static void daemonize(void) { rv = dup2(null_wr, STDERR_FILENO); if (rv < 0) fail("dup2 (stderr) failed\n"); - close(null_rd); - close(null_wr); + if (srv_config.log_file == NULL) close(null_wr); +} + +static void *_ecp_receiver(void *arg) { + ecp_receiver((ECPSocket *)arg); + return NULL; +} + +static int start_receivers(ECPSocket *sock) { + int i, rv; + + for (i=0; i<srv_config.rcvr_thd_num; i++) { + rv = pthread_create(&rcvr_thd[i], NULL, _ecp_receiver, sock); + if (rv) return ECP_ERR; + } + return ECP_OK; +} + +static void dir_result(ECPSocket *sock, ECPDirList *list, int rv) { + unsigned int dir_cnt; + int i; + + if (rv) fail("Fetching directory listing failed err:%d\n", rv); + + dir_cnt = 0; + for (i=0; i<list->count; i++) { + if (list->items[i].roles & ECP_ROLE_DIR) dir_cnt++; + } + rv = dir_init_dir_cnt(dir_cnt, list->serial); + if (rv) fail("Bad timing\n"); + + for (i=0; i<list->count; i++) { + dir_process_item(&list->items[i], sock, NULL); + } + ecp_dir_list_destroy(list); + + pthread_mutex_lock(&dir_sync_mutex); + dir_sync_res = 1; + pthread_cond_signal(&dir_sync_cond); + pthread_mutex_unlock(&dir_sync_mutex); } int main(int argc, char *argv[]) { char *endptr; - char *log_fn; - uid_t uid; - gid_t gid; - int detach; + unsigned int rcvr_num; char *init_switch; - ecp_tr_addr_t my_addr; ECPNode node; ECPContext ctx; ECPSocket sock; + ECPConnection dir_conn; ECPConnHandler vconn_handler; ECPConnHandler vlink_handler; + ECPConnHandler dir_handler; int _argi, _argc, fd; int rv; memset(&srv_config, 0, sizeof(srv_config)); memset(&node, 0, sizeof(node)); - if (argc < 1) fail("Bad arguments\n"); + if (argc < 1) fail("Bad argument count\n"); _argc = argc - 1; _argi = 1; - log_fn = NULL; - detach = 0; - uid = 0; - gid = 0; + srv_config.rcvr_thd_num = 1; while (_argc && (argv[_argi][0] == '-')) { switch (argv[_argi][1]) { case 'd': { _argi++; _argc--; - detach = 1; + srv_config.detach = 1; + break; + } + + case 't': { + _argi++; + _argc--; + if (_argc == 0) usage(argv[0]); + srv_config.rcvr_thd_num = (unsigned int)strtol(argv[_argi], &endptr, 10); + if (endptr[0] != '\0') fail("Bad number of receiver threads: %s\n", argv[_argi]); + if (srv_config.rcvr_thd_num > MAX_THREADS) fail ("Maximum number of threads allowed: %d\n", MAX_THREADS); + _argi++; + _argc--; break; } @@ -289,9 +353,9 @@ int main(int argc, char *argv[]) { _argi++; _argc--; if (_argc == 0) usage(argv[0]); - log_fn = strdup(argv[_argi]); log_level = (uint8_t)strtol(argv[_argi], &endptr, 10); - if ((endptr[0] != '\0') || log_level > LOG_MAX_LEVEL) fail("Bad log level\n"); + if (endptr[0] != '\0') fail("Bad log level: %s\n", argv[_argi]); + if (log_level > LOG_MAX_LEVEL) fail("Maximum log level allowed: %d\n", LOG_MAX_LEVEL); _argi++; _argc--; break; @@ -301,7 +365,7 @@ int main(int argc, char *argv[]) { _argi++; _argc--; if (_argc == 0) usage(argv[0]); - log_fn = strdup(argv[_argi]); + srv_config.log_fn = strdup(argv[_argi]); _argi++; _argc--; break; @@ -311,8 +375,8 @@ int main(int argc, char *argv[]) { _argi++; _argc--; if (_argc == 0) usage(argv[0]); - uid = (uid_t)strtol(argv[_argi], &endptr, 10); - if (endptr[0] != '\0') fail("Bad uid\n"); + srv_config.uid = (uid_t)strtol(argv[_argi], &endptr, 10); + if (endptr[0] != '\0') fail("Bad uid: %s\n", argv[_argi]); _argi++; _argc--; break; @@ -322,8 +386,8 @@ int main(int argc, char *argv[]) { _argi++; _argc--; if (_argc == 0) usage(argv[0]); - gid = (gid_t)strtol(argv[_argi], &endptr, 10); - if (endptr[0] != '\0') fail("Bad gid\n"); + srv_config.gid = (gid_t)strtol(argv[_argi], &endptr, 10); + if (endptr[0] != '\0') fail("Bad gid: %s\n", argv[_argi]); _argi++; _argc--; break; @@ -337,28 +401,28 @@ int main(int argc, char *argv[]) { if (_argc < 4) usage(argv[0]); srv_config.region = (uint8_t)strtol(argv[_argi], &endptr, 16); - if (endptr[0] != '\0') fail("Bad region\n"); - if (srv_config.region >= MAX_REGION) fail("Bad region\n"); + if (endptr[0] != '\0') fail("Bad region: %s\n", argv[_argi]); + if (srv_config.region >= MAX_REGION) fail("Bad region: %d\n", srv_config.region); _argi++; _argc--; srv_config.roles = (uint8_t)strtol(argv[_argi], &endptr, 16); - if (endptr[0] != '\0') fail("Bad roles\n"); + if (endptr[0] != '\0') fail("Bad roles: %s\n", argv[_argi]); _argi++; _argc--; fd = open(argv[_argi], O_RDONLY); - if (fd < 0) fail("Unable to open %s\n", argv[_argi]); + if (fd < 0) fail("Unable to open key file: %s\n", argv[_argi]); - rv = ecp_util_read_key(fd, &srv_config.key_perma.public, &srv_config.key_perma.private); + rv = ecp_util_read_key(fd, &srv_config.key_perma.public, &srv_config.key_perma.private, NULL); close(fd); - if (rv) fail("Unable to read key from %s\n", argv[_argi]); + if (rv) fail("Bad key file: %s\n", argv[_argi]); srv_config.key_perma.valid = 1; _argi++; _argc--; - rv = ecp_addr_init(&my_addr, argv[_argi]); - if (rv) fail("ecp_addr_init err:%d\n", rv); + rv = ecp_addr_init(&srv_config.my_addr, argv[_argi]); + if (rv) fail("Bad local address: %s\n", argv[_argi]); _argi++; _argc--; @@ -376,25 +440,25 @@ int main(int argc, char *argv[]) { _argc--; rv = acl_load(); - if (rv) fail("acl_load err:%d\n", rv); + if (rv) fail("ACL load failed\n"); } if (_argc == 2) { ecp_ecdh_public_t node_pub; fd = open(argv[_argi], O_RDONLY); - if (fd < 0) fail("Unable to open %s\n", argv[_argi]); + if (fd < 0) fail("Unable to open public key file: %s\n", argv[_argi]); - rv = ecp_util_read_key(fd, &node_pub, NULL); + rv = ecp_util_read_key(fd, &node_pub, NULL, NULL); close(fd); - if (rv) fail("Unable to read public key from %s\n", argv[_argi]); + if (rv) fail("Bad public key file: %s\n", argv[_argi]); _argi++; _argc--; ecp_node_init(&node, &node_pub, NULL); rv = ecp_node_set_addr(&node, argv[_argi]); - if (rv) fail("ecp_node_set_addr err:%d\n", rv); + if (rv) fail("Bad remote address: %s\n", argv[_argi]); _argi++; _argc--; } else if (_argc) { @@ -402,12 +466,12 @@ int main(int argc, char *argv[]) { } umask(077); /* rw for owner only */ - if (log_fn) { - log_file = fopen(log_fn, "a"); - if (log_file == NULL) fail("Unable to open log file: %s\n", log_fn); + if (srv_config.log_fn) { + srv_config.log_file = fopen(srv_config.log_fn, "a"); + if (srv_config.log_file == NULL) fail("Unable to open log file: %s\n", srv_config.log_fn); } - if (detach) daemonize(); + if (srv_config.detach) daemonize(); rv = sig_init(); if (rv) fail("sig_init err:%d\n", rv); @@ -424,17 +488,16 @@ int main(int argc, char *argv[]) { rv = ecp_vconn_sock_create(&sock); if (rv) fail("ecp_vconn_sock_create err:%d\n", rv); - rv = ecp_sock_open(&sock, &my_addr); + rv = ecp_sock_open(&sock, &srv_config.my_addr); if (rv) fail("ecp_sock_open err:%d\n", rv); - srv_config.my_addr = my_addr; - if (uid || gid) { - if (gid) { - rv = setgid(gid); + if (srv_config.uid || srv_config.gid) { + if (srv_config.gid) { + rv = setgid(srv_config.gid); if (rv) fail("Unable to set group id\n"); } - if (uid) { - rv = setuid(uid); + if (srv_config.uid) { + rv = setuid(srv_config.uid); if (rv) fail("Unable to set user id\n"); } } @@ -445,24 +508,39 @@ int main(int argc, char *argv[]) { rv = vlink_init(&sock); if (rv) fail("vlink_init err:%d\n", rv); - rv = ecp_start_receiver(&sock); - if (rv) fail("ecp_start_receiver err:%d\n", rv); + rv = start_receivers(&sock); + if (rv) fail("start_receivers err:%d\n", rv); + init_switch = getenv("ECP_INITSW"); if (node.key_perma.valid) { - rv = dir_init_ann(&sock, &node); - if (rv) fail("dir_init_ann err:%d\n", rv); + if (init_switch == NULL) { + rv = ecp_dir_set_handler(&ctx, &dir_handler, dir_result); + if (rv) fail("ecp_dir_set_handler err:%d\n", rv); + + rv = ecp_dir_get(&dir_conn, &sock, &node, 0); + if (rv) fail("ecp_dir_get err:%d\n", rv); + + pthread_mutex_lock(&dir_sync_mutex); + while (dir_sync_res == 0) { + pthread_cond_wait(&dir_sync_cond, &dir_sync_mutex); + } + pthread_mutex_unlock(&dir_sync_mutex); + LOG(LOG_DEBUG, "ecp_dir_get: done\n"); + } else { + rv = dir_init_ann(&sock, &node); + if (rv) fail("dir_init_ann err:%d\n", rv); + } } - init_switch = getenv("ECP_INITSW"); if (init_switch) { int init_ann; init_ann = (int)strtol(init_switch, &endptr, 10); - if (endptr[0] != '\0') fail("Bad environment"); + if (endptr[0] != '\0') fail("Bad environment: ECP_INITSW:%s\n", init_switch); - LOG(LOG_DEBUG, "init switch start - number of announces:%d\n", init_ann); + LOG(LOG_DEBUG, "dir_init_switch: start - number of announces:%d\n", init_ann); dir_init_switch(&sock, init_ann); - LOG(LOG_DEBUG, "init switch done\n"); + LOG(LOG_DEBUG, "dir_init_switch: done\n"); } rv = dir_start_announce(&sock); diff --git a/ecp/server/server.h b/ecp/server/server.h index 3c720ba..c17fa62 100644 --- a/ecp/server/server.h +++ b/ecp/server/server.h @@ -5,6 +5,8 @@ #define LOG_MAX_LEVEL 2 #define LOG(l, ...) { log_printf(l, __VA_ARGS__); } +#define MAX_THREADS 128 + typedef struct SRVConfig { ECPDHKey key_perma; char *acl_fn; @@ -12,8 +14,16 @@ typedef struct SRVConfig { ecp_tr_addr_t my_addr; uint8_t region; uint8_t roles; + int detach; + pthread_t rcvr_thd[MAX_THREADS]; + unsigned int rcvr_thd_num; + uid_t uid; + gid_t gid; + int log_level; + char *log_fn; + FILE *log_file; } SRVConfig; SRVConfig *srv_get_config(void); -void log_vfprintf(int level, FILE *file, const char *format, va_list ap); +void log_vprintf(int level, const char *format, va_list ap); void log_printf(int level, const char *format, ...); diff --git a/ecp/src/ecp/dir/dir_client.c b/ecp/src/ecp/dir/dir_client.c index 3b0d39d..defc202 100644 --- a/ecp/src/ecp/dir/dir_client.c +++ b/ecp/src/ecp/dir/dir_client.c @@ -49,7 +49,7 @@ static ssize_t dir_handle_msg(ECPConnection *conn, ecp_seq_t seq, unsigned char unsigned char frag_tot; int is_first, is_last; int i; - int rv; + int rv = ECP_OK; dir_list_fin = NULL; @@ -99,7 +99,7 @@ handle_dir_msg_fin: ecp_frag_end(conn, seq, frag_cnt, rv); if (rv || is_last) { - if (dir_result_f) dir_result_f(dir_list_fin, rv); + if (dir_result_f) dir_result_f(conn->sock, dir_list_fin, rv); ecp_conn_close(conn); } @@ -108,7 +108,7 @@ handle_dir_msg_fin: } static void dir_handle_err(ECPConnection *conn, unsigned char mtype, int err) { - if (dir_result_f) dir_result_f(NULL, err); + if (dir_result_f) dir_result_f(conn->sock, NULL, err); ecp_conn_close(conn); } diff --git a/ecp/src/ecp/dir/dir_client.h b/ecp/src/ecp/dir/dir_client.h index 1977e95..d55b8f6 100644 --- a/ecp/src/ecp/dir/dir_client.h +++ b/ecp/src/ecp/dir/dir_client.h @@ -5,7 +5,7 @@ typedef struct ECPDirList { ECPDirItem *items; } ECPDirList; -typedef void (*ecp_dir_result_t) (ECPDirList *dir_list, int err); +typedef void (*ecp_dir_result_t) (ECPSocket *sock, ECPDirList *dir_list, int err); ECPDirList *ecp_dir_list_create(void); void ecp_dir_list_destroy(ECPDirList *dir_list); |