/* (c) Copyright 2012 - 2018 Anton Sviridenko https://picapica.im 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 3 of the License, or (at your option) any later version. 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. You should have received a copy of the GNU General Public License along with this program. If not, see . */ #include #include #include #include #include #include #include #include "PICA_security.h" #include "PICA_node.h" #include "PICA_proto.h" #include "PICA_msgproc.h" #include "PICA_nodeaddrlist.h" #include "PICA_netconf.h" #include "PICA_common.h" #include "PICA_nodeconfig.h" #include "PICA_log.h" #include "PICA_nodewait.h" #include "PICA_signverify.h" #ifdef NO_RAND_DEV #include "PICA_rand_seed.h" #endif #ifdef WIN32 #include #ifdef __MINGW32__ #include #endif typedef u_long in_addr_t; typedef u_short in_port_t; typedef u_short uint16_t; #endif #define MAX_NEWCONNS 64 #define NEWCONN_TIMEOUT 10 //sec #define SKYNET_REFRESH_TIMEOUT 93 #define SELECT_TIMEOUT 1 #define NOREPLY_TIMEOUT 15 #define KEEPALIVE_TIMEOUT 60 time_t skynet_refresh_tmst; clock_t TMO_CCLINK_WAITACTIVE = 15; //CONF char *my_addr;//TEMP CONF FIXME static struct event_base *pica_event_base; SOCKET listen_comm_sck;//* static int stop_node_loop = 0; SSL_CTX *ctx; SSL_CTX *anon_ctx; struct newconn newconns[MAX_NEWCONNS]; unsigned int newconns_pos = 0; struct client *client_tree_root; struct client *client_list_head; struct client *client_list_end; struct cclink *cclink_list_head; struct cclink *cclink_list_end; unsigned int cclink_list_count; struct nodelink *nodelink_list_head; struct nodelink *nodelink_list_end; unsigned int nodelink_list_count; const char msg_INITRESPOK[] = {PICA_PROTO_INITRESPOK, PICA_PROTO_INITRESPOK, 'O', 'K'}; const char msg_VERDIFFER[] = {PICA_PROTO_VERDIFFER, PICA_PROTO_VERDIFFER, PICA_PROTO_VER_HIGH, PICA_PROTO_VER_LOW}; unsigned int procmsg_INITREQ(unsigned char* buf, unsigned int size, void* ptr); unsigned int procmsg_CONNID(unsigned char* buf, unsigned int size, void* ptr); unsigned int procmsg_NODECONNREQ(unsigned char* buf, unsigned int size, void* ptr); unsigned int procmsg_CONNREQOUTG(unsigned char* buf, unsigned int size, void* ptr); unsigned int procmsg_CONNALLOW(unsigned char* buf, unsigned int size, void* ptr); unsigned int procmsg_CONNDENY(unsigned char* buf, unsigned int size, void* ptr); unsigned int procmsg_NODERESP(unsigned char* buf, unsigned int size, void* ptr); unsigned int procmsg_NEWNODE(unsigned char* buf, unsigned int size, void* ptr); unsigned int procmsg_NODELISTREQ(unsigned char* buf, unsigned int size, void* ptr); unsigned int procmsg_NODELIST(unsigned char* buf, unsigned int size, void* ptr); unsigned int procmsg_N2NFOUND(unsigned char* buf, unsigned int size, void* ptr); unsigned int procmsg_SEARCH(unsigned char* buf, unsigned int size, void* ptr); unsigned int procmsg_N2NCONNREQOUTG(unsigned char* buf, unsigned int size, void* ptr); unsigned int procmsg_N2NALLOW(unsigned char* buf, unsigned int size, void* ptr); unsigned int procmsg_N2NNOTFOUND(unsigned char* buf, unsigned int size, void* ptr); unsigned int procmsg_N2NMSG(unsigned char* buf, unsigned int size, void* ptr); unsigned int procmsg_PINGREQ_client(unsigned char* buf, unsigned int size, void* ptr); unsigned int procmsg_PINGREP_client(unsigned char* buf, unsigned int size, void* ptr); unsigned int procmsg_PINGREQ_node(unsigned char* buf, unsigned int size, void* ptr); unsigned int procmsg_PINGREP_node(unsigned char* buf, unsigned int size, void* ptr); unsigned int procmsg_dummy_multi_c2n(unsigned char* buf, unsigned int size, void* ptr); unsigned int procmsg_MULTILOGIN_client(unsigned char* buf, unsigned int size, void* ptr); unsigned int procmsg_MULTILOGIN_node(unsigned char* buf, unsigned int size, void* ptr); struct cclink* cclink_list_add(struct client *clr, struct client *cle); struct cclink* cclink_list_search(const unsigned char *caller_id, const unsigned char *callee_id); struct cclink* cclink_list_findwaitsearch(const unsigned char *callee_id); void cclink_list_addlocal(struct client *clr, struct client *cle); void cclink_list_delete(struct cclink *l); void cclink_list_addn2nclr(struct client *clr, const unsigned char *callee_id); void cclink_setwaitconn(struct cclink *ccl); void cclink_list_addn2ncle(const unsigned char *caller_id, struct nodelink *caller_node, struct client *cle); void cclink_activate(struct cclink *ccl); void cclink_setwaitanontlsshutdown(struct cclink *link, struct newconn *nc); void cclink_attach_remotecle_node(struct cclink *ccl, struct nodelink *callee_node); void cclink_list_delete_by_client(struct client *cl); void cclink_list_delete_by_nodelink(struct nodelink *node); struct client* client_tree_search(const unsigned char *id);// struct client *client_list_addnew(struct newconn *nc); void client_list_delete(struct client* ci); struct PICA_proto_msg* client_wbuf_push(struct client *c, unsigned int msgid, unsigned int size); int client_rbuf_grow(struct client *c); struct PICA_proto_msg* nodelink_wbuf_push(struct nodelink *nl, unsigned int msgid, unsigned int size); int nodelink_rbuf_grow(struct nodelink *nl); int nodelink_attach_nodeaddr(struct nodelink *nl, unsigned int addr_type, void* nodeaddr, unsigned int nodeaddr_size); struct nodelink *nodelink_list_addnew(struct newconn *nc); void nodelink_list_delete(struct nodelink *n); struct nodelink *nodelink_search_by_ipv4addr(in_addr_t addr, in_port_t port); /* arguments are in network byte order */ void newconn_close(struct newconn* nc); void newconn_free(struct newconn* nc); struct newconn* newconn_add(struct newconn *ncs, int *pos); void newconns_init(); void PICA_node_joinskynet(const char* addrlistfilename, const char *my_addr); void process_listen_read(evutil_socket_t listener, short event, void *arg); void process_timeout_event(evutil_socket_t listener, short event, void *arg); void process_newconn_read(evutil_socket_t s, short event, void *arg); void process_c2n_read(evutil_socket_t s, short event, void *arg); void process_c2n_write(evutil_socket_t s, short event, void *arg); void process_n2n_read(evutil_socket_t s, short event, void *arg); void process_n2n_write(evutil_socket_t s, short event, void *arg); void process_c2c_read(evutil_socket_t s, short event, void *arg); void process_c2c_write(evutil_socket_t s, short event, void *arg); void process_nodewait(); #define MSGINFO_MSGSNUM(arr) (sizeof(arr)/sizeof(struct PICA_msginfo)) struct PICA_msginfo _msginfo_comm[] = { {PICA_PROTO_CONNREQOUTG, PICA_MSG_FIXED_SIZE, PICA_PROTO_CONNREQOUTG_SIZE, procmsg_CONNREQOUTG}, {PICA_PROTO_CONNALLOW, PICA_MSG_FIXED_SIZE, PICA_PROTO_CONNALLOW_SIZE, procmsg_CONNALLOW}, {PICA_PROTO_CONNDENY, PICA_MSG_FIXED_SIZE, PICA_PROTO_CONNDENY_SIZE, procmsg_CONNDENY}, {PICA_PROTO_CLNODELISTREQ, PICA_MSG_FIXED_SIZE, PICA_PROTO_CLNODELISTREQ_SIZE, procmsg_NODELISTREQ}, {PICA_PROTO_PINGREQ, PICA_MSG_FIXED_SIZE, PICA_PROTO_PINGREQ_SIZE, procmsg_PINGREQ_client}, {PICA_PROTO_PINGREP, PICA_MSG_FIXED_SIZE, PICA_PROTO_PINGREP_SIZE, procmsg_PINGREP_client}, {PICA_PROTO_MULTILOGIN, PICA_MSG_VAR_SIZE, PICA_MSG_VARSIZE_INT16, procmsg_MULTILOGIN_client} }; struct PICA_msginfo _msginfo_node[] = { {PICA_PROTO_INITRESPOK, PICA_MSG_FIXED_SIZE, PICA_PROTO_INITRESPOK_SIZE, procmsg_NODERESP}, {PICA_PROTO_VERDIFFER, PICA_MSG_FIXED_SIZE, PICA_PROTO_VERDIFFER_SIZE, procmsg_NODERESP}, {PICA_PROTO_NEWNODE_IPV4, PICA_MSG_FIXED_SIZE, PICA_PROTO_NEWNODE_IPV4_SIZE, procmsg_NEWNODE}, {PICA_PROTO_NODELISTREQ, PICA_MSG_FIXED_SIZE, PICA_PROTO_NODELISTREQ_SIZE, procmsg_NODELISTREQ}, {PICA_PROTO_NODELIST, PICA_MSG_VAR_SIZE, PICA_MSG_VARSIZE_INT16, procmsg_NODELIST}, {PICA_PROTO_SEARCH, PICA_MSG_FIXED_SIZE, PICA_PROTO_SEARCH_SIZE, procmsg_SEARCH}, {PICA_PROTO_N2NFOUND, PICA_MSG_FIXED_SIZE, PICA_PROTO_N2NFOUND_SIZE, procmsg_N2NFOUND}, {PICA_PROTO_N2NFOUNDCACHE, PICA_MSG_VAR_SIZE, PICA_MSG_VARSIZE_INT16, procmsg_N2NFOUND}, {PICA_PROTO_N2NCONNREQOUTG, PICA_MSG_FIXED_SIZE, PICA_PROTO_N2NCONNREQOUTG_SIZE, procmsg_N2NCONNREQOUTG}, {PICA_PROTO_N2NALLOW, PICA_MSG_FIXED_SIZE, PICA_PROTO_N2NALLOW_SIZE, procmsg_N2NALLOW}, {PICA_PROTO_N2NNOTFOUND, PICA_MSG_FIXED_SIZE, PICA_PROTO_N2NNOTFOUND_SIZE, procmsg_N2NNOTFOUND}, {PICA_PROTO_N2NMSG, PICA_MSG_VAR_SIZE, PICA_MSG_VARSIZE_INT16, procmsg_N2NMSG}, {PICA_PROTO_PINGREQ, PICA_MSG_FIXED_SIZE, PICA_PROTO_PINGREQ_SIZE, procmsg_PINGREQ_node}, {PICA_PROTO_PINGREP, PICA_MSG_FIXED_SIZE, PICA_PROTO_PINGREP_SIZE, procmsg_PINGREP_node}, {PICA_PROTO_MULTILOGIN, PICA_MSG_VAR_SIZE, PICA_MSG_VARSIZE_INT16, procmsg_MULTILOGIN_node} }; struct PICA_msginfo _msginfo_newconn[] = { {PICA_PROTO_INITREQ, PICA_MSG_FIXED_SIZE, PICA_PROTO_INITREQ_SIZE, procmsg_INITREQ}, {PICA_PROTO_CONNID, PICA_MSG_FIXED_SIZE, PICA_PROTO_CONNID_SIZE, procmsg_CONNID}, {PICA_PROTO_NODECONNREQ, PICA_MSG_FIXED_SIZE, PICA_PROTO_NODECONNREQ_SIZE, procmsg_NODECONNREQ} }; struct PICA_msginfo _msginfo_multi[] = { {PICA_PROTO_CONNREQOUTG, PICA_MSG_FIXED_SIZE, PICA_PROTO_CONNREQOUTG_SIZE, procmsg_dummy_multi_c2n}, {PICA_PROTO_CONNALLOW, PICA_MSG_FIXED_SIZE, PICA_PROTO_CONNALLOW_SIZE, procmsg_dummy_multi_c2n}, {PICA_PROTO_CONNDENY, PICA_MSG_FIXED_SIZE, PICA_PROTO_CONNDENY_SIZE, procmsg_dummy_multi_c2n}, {PICA_PROTO_CLNODELISTREQ, PICA_MSG_FIXED_SIZE, PICA_PROTO_CLNODELISTREQ_SIZE, procmsg_NODELISTREQ}, {PICA_PROTO_PINGREQ, PICA_MSG_FIXED_SIZE, PICA_PROTO_PINGREQ_SIZE, procmsg_PINGREQ_client}, {PICA_PROTO_PINGREP, PICA_MSG_FIXED_SIZE, PICA_PROTO_PINGREP_SIZE, procmsg_PINGREP_client}, {PICA_PROTO_MULTILOGIN, PICA_MSG_VAR_SIZE, PICA_MSG_VARSIZE_INT16, procmsg_MULTILOGIN_client} }; static int process_async_ssl_readwrite(SSL *ssl, int ret) { if (ret == 0) return 0; if (ret < 0) { switch(SSL_get_error(ssl, ret)) { case SSL_ERROR_WANT_WRITE: case SSL_ERROR_WANT_READ: break; default: return 0; } } return 1; } static int verify_callback(int preverify_ok, X509_STORE_CTX *ctx) { //return 1 for self-signed certificates char buf[256]; X509 *err_cert; int err, depth; err_cert = X509_STORE_CTX_get_current_cert(ctx); err = X509_STORE_CTX_get_error(ctx); depth = X509_STORE_CTX_get_error_depth(ctx); X509_NAME_oneline(X509_get_subject_name(err_cert), buf, 256); if (err == X509_V_ERR_DEPTH_ZERO_SELF_SIGNED_CERT && depth == 0) preverify_ok = 1; if (!preverify_ok) { PICA_info("certificate verification error:num=%d:%s:depth=%d:%s\n", err, X509_verify_cert_error_string(err), depth, buf); } return preverify_ok; } unsigned int procmsg_INITREQ(unsigned char* buf, unsigned int size, void* ptr) { struct newconn *nc = (struct newconn*)ptr; struct PICA_proto_msg *mp; PICA_debug1("received INITREQ"); nc->type = NEWCONN_C2N; if (buf[2] == PICA_PROTO_VER_HIGH && buf[3] == PICA_PROTO_VER_LOW) { struct client *c; c = client_list_addnew(nc); if (!c) return 0; c->state = PICA_CLSTATE_WAITANONTLSSHUTDOWN; nc->iconn.cl = c; nc->state = PICA_NEWCONN_SENDINGOK; /* newconn_free(nc); if ((mp = client_wbuf_push(c, PICA_PROTO_INITRESPOK, PICA_PROTO_INITRESPOK_SIZE))) { mp->tail[0] = 'O'; mp->tail[1] = 'K'; } c->state = PICA_CLSTATE_SENDINGRESP;*/ }///////---------<<<<<<<<<<< new client else { nc->state = PICA_NEWCONN_SENDINGREJECT; if (buf[2] > PICA_PROTO_VER_HIGH || (buf[2] == PICA_PROTO_VER_HIGH && buf[3] > PICA_PROTO_VER_LOW)) { PICA_warn("Client has newer protocol version than node. Please check if update for pica-node is available"); } else { PICA_info("Client has older protocol version. Disconnecting ..."); } } return 1; } void cclink_sendcleconnected(struct cclink *link) { struct PICA_proto_msg *mp; if (link->state == PICA_CCLINK_LOCAL_WAITCONNCLEANONTLSSHUTDOWN) { if((mp = client_wbuf_push(link->p1, PICA_PROTO_FOUND, PICA_PROTO_FOUND_SIZE))) { memcpy(mp->tail, link->callee_id, PICA_ID_SIZE); } else { cclink_list_delete(link); return; } } if (link->state == PICA_CCLINK_N2NCLE_WAITCONNCLEANONTLSSHUTDOWN) { if((mp = nodelink_wbuf_push(link->caller_node, PICA_PROTO_N2NALLOW, PICA_PROTO_N2NALLOW_SIZE))) { memcpy(mp->tail, link->caller_id, PICA_ID_SIZE); memcpy(mp->tail + PICA_ID_SIZE, link->callee_id, PICA_ID_SIZE); } else { cclink_list_delete(link); return; } } } unsigned int procmsg_CONNID(unsigned char* buf, unsigned int size, void* ptr) { struct PICA_proto_msg *mp; struct newconn *nc = (struct newconn*)ptr; unsigned char *caller_id, *callee_id; struct cclink *link; PICA_debug1("received CONNID"); nc->type = NEWCONN_C2C; caller_id = buf + 2; callee_id = buf + 2 + PICA_ID_SIZE; { char caller_id_buf[2 * PICA_ID_SIZE], callee_id_buf[2 * PICA_ID_SIZE]; PICA_debug1("CONNID: caller_id=%s callee_id=%s", PICA_id_to_base64(caller_id, caller_id_buf), PICA_id_to_base64(callee_id, callee_id_buf)); } link = cclink_list_search(caller_id, callee_id); if (!link) { PICA_debug3("corresponding cclink not found"); return 0; } if (link->state != PICA_CCLINK_LOCAL_WAITCONNCLE && link->state != PICA_CCLINK_LOCAL_WAITCONNCLR && link->state != PICA_CCLINK_N2NCLR_WAITCONNCLR && link->state != PICA_CCLINK_N2NCLE_WAITCONNCLE) { PICA_debug2("found cclink is not in expected state"); return 0; } { struct client *c; //проверка IP адреса. //IP адрес должен совпадать с адресом клиента в управляющем соединении if (link->state == PICA_CCLINK_LOCAL_WAITCONNCLE || link->state == PICA_CCLINK_N2NCLE_WAITCONNCLE) c = link->p2; else if (link->state == PICA_CCLINK_LOCAL_WAITCONNCLR || link->state == PICA_CCLINK_N2NCLR_WAITCONNCLR) c = link->p1; if (nc->addr.sin_addr.s_addr != c->addr.sin_addr.s_addr) { PICA_warn("CONNID: IP check failed. IP address of sender does not match IP in c2n connection"); return 0; } } PICA_debug3("CONNID processed successfully"); cclink_setwaitanontlsshutdown(link, nc); return 1; } unsigned int procmsg_NODECONNREQ(unsigned char* buf, unsigned int size, void* ptr) { struct newconn *nc = (struct newconn*)ptr; struct PICA_proto_msg *mp; PICA_debug1("received NODECONNREQ"); nc->type = NEWCONN_N2N; if (buf[2] == PICA_PROTO_VER_HIGH && buf[3] == PICA_PROTO_VER_LOW) { ////-прием входящего подключения //-------------------------- nc->state = PICA_NEWCONN_SENDINGOK; /*struct nodelink *nl; nl = nodelink_list_addnew(nc); newconn_free(nc); if (!nl) return 0; if ((mp = nodelink_wbuf_push(nl, PICA_PROTO_INITRESPOK, PICA_PROTO_INITRESPOK_SIZE))) { mp->tail[0] = 'O'; mp->tail[1] = 'K'; }*/ } else { nc->state = PICA_NEWCONN_SENDINGREJECT; /* //// PICA_PROTO_VERDIFFER send(nc->sck, msg_VERDIFFER, msg_VERDIFFER_len, 0); if (buf[2] > PICA_PROTO_VER_HIGH || (buf[2] == PICA_PROTO_VER_HIGH && buf[3] > PICA_PROTO_VER_LOW)) { PICA_warn("Node %s has newer protocol version [%u, %u]. Please check if update for pica-node is available", inet_ntoa(nc->addr.sin_addr), buf[2], buf[3]); } else { PICA_info("Node %s has older protocol version. Disconnecting ...", inet_ntoa(nc->addr.sin_addr)); } return 0;*/ } return 1; } void mapfunc_sendmultilogin(struct nodelink *p, const unsigned char *msgbuf) { struct PICA_proto_msg *mp; uint16_t payload_size = *(uint16_t*)(msgbuf + 2); mp = nodelink_wbuf_push(p, PICA_PROTO_MULTILOGIN, payload_size + 4); if (mp) memcpy(mp->tail, msgbuf + 2, payload_size + 2); } void mapfunc_sendsearchreq(struct nodelink *p, const unsigned char *id) { struct PICA_proto_msg *mp; mp = nodelink_wbuf_push(p, PICA_PROTO_SEARCH, PICA_PROTO_SEARCH_SIZE); if (mp) memcpy(mp->tail, id, PICA_ID_SIZE); } int nodelink_send_searchreq(const unsigned char *id) { if (!nodelink_list_count) return 0; LISTMAP(nodelink, nodelink, mapfunc_sendsearchreq, id); return 1; } unsigned int procmsg_CONNREQOUTG(unsigned char* buf, unsigned int size, void* ptr) { struct client *i, *o; //i - указатель на вызывающего клиента, o - вызываемого(если найден) unsigned char *callee_id; struct PICA_proto_msg *mp; i = (struct client *)ptr; PICA_debug1("received CONNREQOUTG from %s", PICA_id_to_base64(i->id, NULL)); callee_id = buf + 2; if ((o = client_tree_search(callee_id))) { if (cclink_list_search(i->id, callee_id)) { PICA_info("duplicate request for existing c2c connection is ignored"); return 1; } if (memcmp(i->id, callee_id, PICA_ID_SIZE) == 0) { PICA_info("request to create c2c connection to self is ignored"); return 1; } //отправить o _CONNREQINC if ((mp = client_wbuf_push(o, PICA_PROTO_CONNREQINC, PICA_PROTO_CONNREQINC_SIZE))) { memcpy(mp->tail, i->id, PICA_ID_SIZE); cclink_list_addlocal(i, o); o->tmst = time(0); o->disconnect_ticking = 1;//if client does not reply in given time period, client is disconnected return 1; } return 1;//сообщение отправить не удалось, но все равно возвращаем 1 } if (cclink_list_findwaitsearch(callee_id)/*search request for this id is already in progress*/ || nodelink_send_searchreq(callee_id) /*make new search*/) { cclink_list_addn2nclr(i, callee_id); return 1; } //отправить i _NOTFOUND if ((mp = client_wbuf_push(i, PICA_PROTO_NOTFOUND, PICA_PROTO_NOTFOUND_SIZE))) { memcpy(mp->tail, callee_id, PICA_ID_SIZE); } return 1; } unsigned int procmsg_CONNALLOW(unsigned char* buf, unsigned int size, void* ptr) { struct client *callee; struct cclink *link; unsigned char *caller_id; PICA_debug1("received CONALLOW"); callee = (struct client *)ptr; caller_id = buf + 2; callee->disconnect_ticking = 0; callee->tmst = time(0); link = cclink_list_search(caller_id, callee->id); if (!link) return 0; if (link->state != PICA_CCLINK_LOCAL_WAITREP && link->state != PICA_CCLINK_N2NCLE_WAITREP) return 0; cclink_setwaitconn(link); return 1; } unsigned int procmsg_CONNDENY(unsigned char* buf, unsigned int size, void* ptr) { struct client *caller, *callee; struct cclink *link; unsigned char *caller_id; struct PICA_proto_msg *mp; PICA_debug1("received CONNDENY"); callee = (struct client *)ptr; caller_id = buf + 2; callee->disconnect_ticking = 0; callee->tmst = time(0); link = cclink_list_search(caller_id, callee->id); if (!link) return 0; caller = link -> p1;//check for NULL if not LOCAL if (link->state != PICA_CCLINK_LOCAL_WAITREP && link->state != PICA_CCLINK_N2NCLE_WAITREP) { cclink_list_delete(link); return 0; } if (link->state == PICA_CCLINK_LOCAL_WAITREP) { //отправить caller _NOTFOUND if ((mp = client_wbuf_push(caller, PICA_PROTO_NOTFOUND, PICA_PROTO_NOTFOUND_SIZE))) { memcpy(mp->tail, callee->id, PICA_ID_SIZE); } } else if (link->state == PICA_CCLINK_N2NCLE_WAITREP) { //отправить caller N2NNOTFOUND if ((mp = nodelink_wbuf_push(link->caller_node, PICA_PROTO_N2NNOTFOUND, PICA_PROTO_N2NNOTFOUND_SIZE))) { memcpy(mp->tail, caller_id, PICA_ID_SIZE); memcpy(mp->tail + PICA_ID_SIZE, callee->id, PICA_ID_SIZE); } } cclink_list_delete(link); return 1; } unsigned int procmsg_PINGREQ_client(unsigned char* buf, unsigned int size, void* ptr) { struct client *c = (struct client*)ptr; struct PICA_proto_msg *mp; PICA_debug3("PINGREQ"); if ((mp = client_wbuf_push(c, PICA_PROTO_PINGREP, PICA_PROTO_PINGREP_SIZE))) { RAND_pseudo_bytes(mp->tail, 2); } else return 0; return 1; } unsigned int procmsg_PINGREP_client(unsigned char* buf, unsigned int size, void* ptr) { struct client *c = (struct client*)ptr; PICA_debug3("PINGREP c2n"); c->disconnect_ticking = 0; c->tmst = time(0); return 1; } unsigned int procmsg_MULTILOGIN_client(unsigned char* buf, unsigned int size, void* ptr) { struct client *other, *me = (struct client*)ptr; PICA_debug3("MULTILOGIN c2n"); //send to other instances connected to this node other = client_tree_search(me->id); while(other) { if (other != me) { struct PICA_proto_msg *mp; if ((mp = client_wbuf_push(other, PICA_PROTO_MULTILOGIN, size))) { memcpy(mp->tail, buf + 2, size - 2); } } other = other->next_multi; } //relay to other nodes LISTMAP(nodelink, nodelink, mapfunc_sendmultilogin, buf); return 1; } unsigned int procmsg_MULTILOGIN_node(unsigned char* buf, unsigned int size, void* ptr) { struct nodelink *n = (struct nodelink *)ptr; struct client *c = client_list_head; uint64_t timestamp; void *sigdatas[4]; size_t sigdatalengths[4]; uint16_t payload_len = *(uint16_t*)(buf + 2); int ret; size_t siglen; unsigned char *sig; PICA_debug3("MULTILOGIN n2n"); if (payload_len <= sizeof timestamp + PICA_PROTO_NODELIST_ITEM_IPV4_SIZE) return 0; timestamp = *(uint64_t*)(buf + 4); //TODO compare timestamps sigdatalengths[0] = PICA_ID_SIZE; //add timestamp to signature data sigdatas[1] = buf + 4; sigdatalengths[1] = sizeof(uint64_t); //add sender node address to signature data sigdatas[2] = buf + 4 + sizeof(uint64_t); switch (buf[4 + sizeof(uint64_t)]) { case PICA_PROTO_NEWNODE_IPV4: sigdatalengths[2] = PICA_PROTO_NODELIST_ITEM_IPV4_SIZE; break; case PICA_PROTO_NEWNODE_IPV6: sigdatalengths[2] = PICA_PROTO_NODELIST_ITEM_IPV6_SIZE; break; case PICA_PROTO_NEWNODE_DNS: sigdatalengths[2] = 4 + buf[5 + sizeof(uint64_t)]; default: PICA_warn("Received MULTILOGIN message with invalid node address"); return 1; } siglen = payload_len - sizeof(uint64_t) - sigdatalengths[2]; sig = buf + 4 + sizeof(uint64_t) + sigdatalengths[2]; sigdatas[3] = NULL; sigdatalengths[3] = 0; while(c) { //check signature against current client ID and data in MULTILOGIN packet if (c->state == PICA_CLSTATE_CONNECTED) { EVP_PKEY *pubkey; X509* x; sigdatas[0] = c->id; if ((x = SSL_get_peer_certificate(c->ssl_comm)) && (pubkey = X509_get_pubkey(x))) { ret = PICA_signverify(pubkey, sigdatas, sigdatalengths, sig, siglen); EVP_PKEY_free(pubkey); X509_free(x); if (ret == 1) { PICA_debug3("found matching client connection for incoming n2n MULTILOGIN message"); break; } } } c = c->next; } if (!c) { PICA_debug3("no matching client found for receved n2n MULTILOGIN"); return 1; } while(c) //relay received MULTILOGIN to each client with same id { struct PICA_proto_msg *mp; mp = client_wbuf_push(c, PICA_PROTO_MULTILOGIN, size); if (mp) { memcpy(mp, buf, size); } c = c->next_multi; } return 1; } unsigned int procmsg_dummy_multi_c2n(unsigned char* buf, unsigned int size, void* ptr) { PICA_debug3("message %X from %p is handled by dummy procmsg", buf[0], ptr); return 1; } unsigned int procmsg_PINGREQ_node(unsigned char* buf, unsigned int size, void* ptr) { struct nodelink *n = (struct nodelink *)ptr; struct PICA_proto_msg *mp; PICA_debug3("PINGREQ"); if ((mp = nodelink_wbuf_push(n, PICA_PROTO_PINGREP, PICA_PROTO_PINGREP_SIZE))) { RAND_pseudo_bytes(mp->tail, 2); } else return 0; return 1; } unsigned int procmsg_PINGREP_node(unsigned char* buf, unsigned int size, void* ptr) { struct nodelink *n = (struct nodelink *)ptr; PICA_debug3("PINGREP n2n"); n->disconnect_ticking = 0; n->tmst = time(0); return 1; } unsigned int procmsg_NODERESP(unsigned char* buf, unsigned int size, void* ptr) { struct nodelink *n = (struct nodelink *)ptr; struct PICA_proto_msg *mp; PICA_debug1("received NODERESP"); switch(buf[0]) { case PICA_PROTO_INITRESPOK: if (buf[2] != 'O' || buf[3] != 'K') return 0; mp = nodelink_wbuf_push(n, PICA_PROTO_NEWNODE_IPV4, PICA_PROTO_NEWNODE_IPV4_SIZE);//IPv6 if (mp) { *((in_addr_t*)(mp->tail)) = inet_addr(nodecfg.announced_addr); *((in_port_t*)(mp->tail + 4)) = (nodecfg.listen_port ? htons(atoi(nodecfg.listen_port)) : htons(PICA_COMM_PORT)); //CONF; } else return 0; mp = nodelink_wbuf_push(n, PICA_PROTO_NODELISTREQ, PICA_PROTO_NODELISTREQ_SIZE); if (mp) { RAND_pseudo_bytes(mp->tail, 2); } else return 0; break; case PICA_PROTO_VERDIFFER: if (buf[2] > PICA_PROTO_VER_HIGH || (buf[3] > PICA_PROTO_VER_LOW && buf[2] == PICA_PROTO_VER_HIGH)) { PICA_warn("make update of node software - peer node has newer protocol version"); } else { //peer node has older software version PICA_info("peer node has older protocol version"); } return 0; break; } return 1; } unsigned int procmsg_NEWNODE(unsigned char* buf, unsigned int size, void* ptr) { struct nodelink *n = (struct nodelink *)ptr; struct PICA_nodeaddr na; PICA_debug1("received NEWNODE"); switch (buf[0]) { case PICA_PROTO_NEWNODE_IPV4: PICA_debug1("NEWNODE_IPV4: ip %.16s port %hu\n", inet_ntoa(*(struct in_addr*)(buf + 2)), ntohs(*(in_port_t*)(buf + 6))); { struct PICA_nodeaddr_ipv4 na_ipv4; na_ipv4.magick = PICA_PROTO_NEWNODE_IPV4; na_ipv4.addr = *(in_addr_t*)(buf + 2); na_ipv4.port = *(in_port_t*)(buf + 6); if (nodelink_search_by_ipv4addr(na_ipv4.addr, na_ipv4.port)) //connection with this node is already established { PICA_debug1("disconnecting node %.16s:%hu - connection with this node is already established", inet_ntoa(*(struct in_addr*)&na_ipv4.addr), ntohs(na_ipv4.port)); return 0; } if (nodecfg.disable_reserved_addrs && PICA_is_reserved_addr_ipv4(na_ipv4.addr)) { PICA_debug1("disconnecting node %.16s:%hu - private IP ranges are rejected by node configuration", inet_ntoa(*(struct in_addr*)&na_ipv4.addr), ntohs(na_ipv4.port) ); return 0; } sprintf(na.addr, "%.16s", inet_ntoa(*(struct in_addr*)&na_ipv4.addr)); na.port = ntohs(na_ipv4.port); //int nodelink_attach_nodeaddr(struct nodelink *nl, unsigned int addr_type, void* nodeaddr, unsigned int nodeaddr_size) nodelink_attach_nodeaddr(n, na_ipv4.magick, &na_ipv4, sizeof(struct PICA_nodeaddr_ipv4)); } break; case PICA_PROTO_NEWNODE_IPV6: #warning "TO DO" break; case PICA_PROTO_NEWNODE_DNS: #warning "TODO" break; } PICA_nodeaddr_save(nodecfg.nodes_db_file, &na);//CONF filename return 1; } unsigned int procmsg_NODELISTREQ(unsigned char* buf, unsigned int size, void* ptr) { struct PICA_proto_msg *mp; struct nodelink *nlp; unsigned int nl_size = 0; struct nodelink *n = 0; struct client *c = 0; unsigned char *nl_buf = 0; PICA_debug1("received NODELISTREQ"); switch(buf[0]) { case PICA_PROTO_CLNODELISTREQ: c = (struct client *)ptr; break; case PICA_PROTO_NODELISTREQ: n = (struct nodelink *)ptr; break; } nl_buf = (unsigned char*)malloc(65536); if (!nl_buf) return 0;//ERR { //TEMP FIXME CONF //собственный адрес узла nl_buf[nl_size] = PICA_PROTO_NEWNODE_IPV4; *((in_addr_t*)(nl_buf + nl_size + 1)) = inet_addr(my_addr); *((in_port_t*)(nl_buf + nl_size + 5)) = (nodecfg.listen_port ? htons(atoi(nodecfg.listen_port)) : htons(PICA_COMM_PORT)); //CONF nl_size += PICA_PROTO_NODELIST_ITEM_IPV4_SIZE; } nlp = nodelink_list_head; while(nlp && nl_size < 65532)//MAP { switch(nlp->addr_type) { case PICA_PROTO_NEWNODE_IPV4: if ( PICA_PROTO_NODELISTREQ == buf[0] && ((struct PICA_nodeaddr_ipv4*)nlp->node_addr)->addr == ((struct PICA_nodeaddr_ipv4*)n->node_addr)->addr && ((struct PICA_nodeaddr_ipv4*)nlp->node_addr)->port == ((struct PICA_nodeaddr_ipv4*)n->node_addr)->port ) break;//exclude requester's address nl_buf[nl_size] = PICA_PROTO_NEWNODE_IPV4; *((in_addr_t*)(nl_buf + nl_size + 1)) = ((struct PICA_nodeaddr_ipv4*)nlp->node_addr)->addr; *((in_port_t*)(nl_buf + nl_size + 5)) = ((struct PICA_nodeaddr_ipv4*)nlp->node_addr)->port; nl_size += PICA_PROTO_NODELIST_ITEM_IPV4_SIZE;//CONF break; } //#warning "set limit on list size, <65532 bytes!!!" nlp = nlp->next; } switch(buf[0]) { case PICA_PROTO_CLNODELISTREQ: mp = client_wbuf_push( c, PICA_PROTO_CLNODELIST, nl_size + 4); break; case PICA_PROTO_NODELISTREQ: mp = nodelink_wbuf_push( n, PICA_PROTO_NODELIST, nl_size + 4); break; } if (mp) { *((uint16_t*)mp->tail) = nl_size; memcpy(mp->tail + 2, nl_buf, nl_size); } free(nl_buf); return 1; } unsigned int procmsg_NODELIST(unsigned char* buf, unsigned int size, void* ptr) { struct PICA_nodeaddr_ipv4 na_ipv4; struct nodelink *n = (struct nodelink *)ptr; unsigned int listsize = size - 2 - PICA_MSG_VARSIZE_INT16; unsigned int listleft = listsize; PICA_debug1("received NODELIST"); while(listleft) { #warning "strict error checking!!! -- listleft overflow, unknown list items, check listleft >= possible item size" if (listleft == listsize) //first entry, sender node's address { switch(buf[size - listleft]) { case PICA_PROTO_NEWNODE_IPV4: na_ipv4.magick = PICA_PROTO_NEWNODE_IPV4; na_ipv4.addr = *(in_addr_t*)(buf + size - listleft + 1); na_ipv4.port = *(in_port_t*)(buf + size - listleft + 5); nodelink_attach_nodeaddr(n, na_ipv4.magick, &na_ipv4, sizeof(struct PICA_nodeaddr_ipv4)); listleft -= PICA_PROTO_NODELIST_ITEM_IPV4_SIZE; break; } continue; } else { struct PICA_nodeaddr na; //char addr[256]; struct newconn nc; struct nodelink *nlp; int skip_save = 0; memset(na.addr, 0, 256); memset(&nc, 0, sizeof(struct newconn)); switch(buf[size - listleft]) { case PICA_PROTO_NEWNODE_IPV4: na_ipv4.magick = PICA_PROTO_NEWNODE_IPV4; na_ipv4.addr = *(in_addr_t*)(buf + size - listleft + 1); na_ipv4.port = *(in_port_t*)(buf + size - listleft + 5); sprintf(na.addr, "%.16s", inet_ntoa(*(struct in_addr*)&na_ipv4.addr)); na.port = ntohs(na_ipv4.port); //nodewait_start_connection(&na); // if (try_connect_to_node(na.addr,ntohs(na_ipv4.port),&nc)) // if (try_get_reply(&nc)) // { // nlp=nodelink_list_addnew(&nc); // send_newnode(nlp,my_addr); // nodelink_attach_nodeaddr(nlp,na_ipv4.magick,&na_ipv4,sizeof(struct PICA_nodeaddr_ipv4)); // PICA_debug1("LINKED NODE:%s\n",inet_ntoa(*(struct in_addr*)&na_ipv4.addr)); // } PICA_debug3("NODELIST item %.16s:%hu", inet_ntoa(*(struct in_addr*)&na_ipv4.addr), ntohs(na_ipv4.port)); if (nodecfg.disable_reserved_addrs && PICA_is_reserved_addr_ipv4(na_ipv4.addr)) { skip_save = 1; PICA_debug1("Received NODELIST item %.16s:%hu is in private range, skipping", inet_ntoa(*(struct in_addr*)&na_ipv4.addr), ntohs(na_ipv4.port) ); } listleft -= PICA_PROTO_NODELIST_ITEM_IPV4_SIZE; break; } if (!skip_save) PICA_nodeaddr_save(nodecfg.nodes_db_file, &na);//CONF filename } } return 1; } unsigned int procmsg_N2NFOUND(unsigned char* buf, unsigned int size, void* ptr) { struct cclink *cc; struct nodelink *n = (struct nodelink *)ptr; unsigned char *callee_id = buf + 2; struct PICA_proto_msg *mp; PICA_debug1("received N2NFOUND"); while((cc = cclink_list_findwaitsearch(callee_id))) { cclink_attach_remotecle_node(cc, n); if((mp = nodelink_wbuf_push(n, PICA_PROTO_N2NCONNREQOUTG, PICA_PROTO_N2NCONNREQOUTG_SIZE))) { memcpy(mp->tail, cc->caller_id, PICA_ID_SIZE); memcpy(mp->tail + PICA_ID_SIZE, callee_id, PICA_ID_SIZE); } } return 1; } unsigned int procmsg_SEARCH(unsigned char* buf, unsigned int size, void* ptr) { struct nodelink *n = (struct nodelink *)ptr; unsigned char *callee_id = buf + 2; struct PICA_proto_msg *mp; struct client *callee; PICA_debug1("received SEARCH"); if ((callee = client_tree_search(callee_id))) { if ((mp = nodelink_wbuf_push(n, PICA_PROTO_N2NFOUND, PICA_PROTO_N2NFOUND_SIZE))) { memcpy(mp->tail, callee_id, PICA_ID_SIZE); } } return 1; } unsigned int procmsg_N2NCONNREQOUTG(unsigned char* buf, unsigned int size, void* ptr) { struct nodelink *n = (struct nodelink *)ptr; unsigned char *caller_id, *callee_id; struct PICA_proto_msg *mp; struct client *callee; PICA_debug1("received N2NCONNREQOUTG"); caller_id = buf + 2; callee_id = buf + 2 + PICA_ID_SIZE; if ((callee = client_tree_search(callee_id))) { if ((mp = client_wbuf_push(callee, PICA_PROTO_CONNREQINC, PICA_PROTO_CONNREQINC_SIZE))) { memcpy(mp->tail, caller_id, PICA_ID_SIZE); cclink_list_addn2ncle(caller_id, n, callee); } } else { if ((mp = nodelink_wbuf_push(n, PICA_PROTO_N2NNOTFOUND, PICA_PROTO_N2NNOTFOUND_SIZE))) { memcpy(mp->tail, caller_id, PICA_ID_SIZE); memcpy(mp->tail + PICA_ID_SIZE, callee_id, PICA_ID_SIZE); } } return 1; } unsigned int procmsg_N2NALLOW(unsigned char* buf, unsigned int size, void* ptr) { struct nodelink *n = (struct nodelink *)ptr; unsigned char *caller_id, *callee_id; struct PICA_proto_msg *mp; struct cclink *cc; PICA_debug1("received N2NALLOW"); caller_id = buf + 2; callee_id = buf + 2 + PICA_ID_SIZE; cc = cclink_list_search(caller_id, callee_id); if (cc) { if ((mp = client_wbuf_push(cc->p1, PICA_PROTO_FOUND, PICA_PROTO_FOUND_SIZE))) { memcpy(mp->tail, callee_id, PICA_ID_SIZE); cclink_setwaitconn(cc); } } return 1; } unsigned int procmsg_N2NNOTFOUND(unsigned char* buf, unsigned int size, void* ptr) { struct nodelink *n = (struct nodelink *)ptr; unsigned char *caller_id, *callee_id; struct PICA_proto_msg *mp; struct cclink *cc; PICA_debug1("received N2NNOTFOUND"); caller_id = buf + 2; callee_id = buf + 2 + PICA_ID_SIZE; cc = cclink_list_search(caller_id, callee_id); if (cc) { if ((mp = client_wbuf_push(cc->p1, PICA_PROTO_NOTFOUND, PICA_PROTO_NOTFOUND_SIZE))) { memcpy(mp->tail, callee_id, PICA_ID_SIZE); } cclink_list_delete(cc); } return 1; } unsigned int procmsg_N2NMSG(unsigned char* buf, unsigned int size, void* ptr) { struct nodelink *n = (struct nodelink *)ptr; struct PICA_proto_msg *mp; unsigned char *sender_id, *receiver_id; struct cclink *cc; unsigned int len, datalen; PICA_debug1("received N2NMSG"); len = *(unsigned short*)(buf + 2); sender_id = buf + 4; receiver_id = buf + 4 + PICA_ID_SIZE; if (len < 2 * PICA_ID_SIZE) return 0; datalen = len - 2 * PICA_ID_SIZE; cc = cclink_list_search(sender_id, receiver_id); if (cc && cc->state == PICA_CCLINK_N2NCLE_ACTIVE) { if (datalen + cc->bufpos_p1p2 > cc->buflen_p1p2) { cclink_list_delete(cc);//CLOSING CCLINK DUE TO BUFER OVERRUN --------- FIXME FIX FIX FIX return 1; } memcpy(cc->buf_p1p2 + cc->bufpos_p1p2, buf + 4 + 2 * PICA_ID_SIZE, datalen); cc->bufpos_p1p2 += datalen; return 1; } /////////////// cc = cclink_list_search(receiver_id, sender_id); if (cc && cc->state == PICA_CCLINK_N2NCLR_ACTIVE) { if (datalen + cc->bufpos_p2p1 > cc->buflen_p2p1) { cclink_list_delete(cc);//CLOSING CCLINK DUE TO BUFER OVERRUN --------- FIXME FIX FIX FIX return 1; } memcpy(cc->buf_p2p1 + cc->bufpos_p2p1, buf + 4 + 2 * PICA_ID_SIZE, datalen); cc->bufpos_p2p1 += datalen; return 1; } if ((mp = nodelink_wbuf_push(n, PICA_PROTO_N2NNOTFOUND, PICA_PROTO_N2NNOTFOUND_SIZE))) { memcpy(mp->tail, sender_id, PICA_ID_SIZE); memcpy(mp->tail + PICA_ID_SIZE, receiver_id, PICA_ID_SIZE); } return 1; } struct cclink* cclink_list_findwaitsearch(const unsigned char *callee_id) { struct cclink *c; c = cclink_list_head; while(c) { if (memcmp(c->callee_id, callee_id, PICA_ID_SIZE) == 0 && c->state == PICA_CCLINK_N2NCLR_WAITSEARCH) return c; c = c->next; } return 0; } struct cclink* cclink_list_search(const unsigned char *caller_id, const unsigned char *callee_id) { struct cclink *c; c = cclink_list_head; while(c) { if (memcmp(c->caller_id, caller_id, PICA_ID_SIZE) == 0 && memcmp(c->callee_id, callee_id, PICA_ID_SIZE) == 0) return c; c = c->next; } return 0; } void cclink_list_delete(struct cclink *l) { PICA_debug1("cclink_list_delete(%p)", (void*)l); //удаление из глобального списка if (l->prev) l->prev->next = l->next; else cclink_list_head = l->next; if (l->next) l->next->prev = l->prev; else cclink_list_end = l->prev; if (l->p1) { if (l->ev_read_p1) { event_del(l->ev_read_p1); event_free(l->ev_read_p1); } if (l->ev_write_p1) { event_del(l->ev_write_p1); event_free(l->ev_write_p1); } if (l->sck_p1 != -1) { SHUTDOWN(l->sck_p1); CLOSE(l->sck_p1); PICA_debug2("closed c2c socket %i", l->sck_p1); } }//caller_id is stored in p1 structure else free(l->caller_id);//caller_id was allocated by malloc if (l->p2) { if (l->ev_read_p2) { event_del(l->ev_read_p2); event_free(l->ev_read_p2); } if (l->ev_write_p2) { event_del(l->ev_write_p2); event_free(l->ev_write_p2); } if (l->sck_p2 != -1) { SHUTDOWN(l->sck_p2); CLOSE(l->sck_p2); PICA_debug2("closed c2c socket %i", l->sck_p2); } } else free(l->callee_id); if (l->buf_p1p2) free(l->buf_p1p2); if (l->buf_p2p1) free(l->buf_p2p1); free(l); cclink_list_count--; } void cclink_list_delete_by_client(struct client *cl) { struct cclink *kill_ptr = 0, *l = cclink_list_head; PICA_debug1("cclink_list_delete_by_client(%p)", (void*)cl); while(l) { if (l->p1 == cl || l->p2 == cl) kill_ptr = l; l = l->next; if (kill_ptr) { cclink_list_delete(kill_ptr); kill_ptr = 0; } } } void cclink_list_delete_by_nodelink(struct nodelink *node) { struct cclink *kill_ptr = 0, *l = cclink_list_head; PICA_debug1("cclink_list_delete_by_nodelink(%p)", (void*)node); while(l) { if (l->caller_node == node || l->callee_node == node) kill_ptr = l; l = l->next; if (kill_ptr) { cclink_list_delete(kill_ptr); kill_ptr = 0; } } } //clr - caller //cle - callee // null value of clr or cle means that corresponding client is on remote node, is not local client struct cclink* cclink_list_add(struct client *clr, struct client *cle) { struct cclink *l; l = (struct cclink*)calloc(sizeof(struct cclink), 1); if (!l) return 0; //l->state=PICA_LINKSTATE_WAITREP; l->p1 = clr; l->p2 = cle; //добавление в глобальный список if (cclink_list_head) { cclink_list_end->next = l; l->prev = cclink_list_end; cclink_list_end = l; } else { cclink_list_head = l; cclink_list_end = l; } //timestamp l->tmst = time(0); cclink_list_count++; l->sck_p1 = -1; l->sck_p2 = -1; return l; } void cclink_list_addlocal(struct client *clr, struct client *cle) { struct cclink *l; l = cclink_list_add(clr, cle); if (l) { l->state = PICA_CCLINK_LOCAL_WAITREP; l->caller_id = l->p1->id; l->callee_id = l->p2->id; } } void cclink_list_addn2nclr(struct client *clr, const unsigned char *callee_id) { struct cclink *l; l = cclink_list_add(clr, 0); if (l) { l->state = PICA_CCLINK_N2NCLR_WAITSEARCH; l->caller_id = l->p1->id; l->callee_id = (unsigned char*)malloc(PICA_ID_SIZE); if (l->callee_id == NULL) PICA_fatal("memory allocation failed"); memcpy(l->callee_id, callee_id, PICA_ID_SIZE); } } void cclink_attach_remotecle_node(struct cclink *ccl, struct nodelink *callee_node) { if (ccl && ccl->state == PICA_CCLINK_N2NCLR_WAITSEARCH) { ccl->state = PICA_CCLINK_N2NCLR_WAITREP; ccl->callee_node = callee_node; } } void cclink_list_addn2ncle(const unsigned char *caller_id, struct nodelink *caller_node, struct client *cle) { struct cclink *l; l = cclink_list_add(0, cle); if (l) { l->state = PICA_CCLINK_N2NCLE_WAITREP; l->caller_id = (unsigned char*)malloc(PICA_ID_SIZE); if (l->caller_id == NULL) PICA_fatal("memory allocation failed"); memcpy(l->caller_id, caller_id, PICA_ID_SIZE); l->callee_id = l->p2->id; l->caller_node = caller_node; } } void cclink_setwaitconn(struct cclink *ccl) { switch (ccl->state) { case PICA_CCLINK_LOCAL_WAITREP: ccl->state = PICA_CCLINK_LOCAL_WAITCONNCLE; // break; case PICA_CCLINK_N2NCLR_WAITREP: ccl->state = PICA_CCLINK_N2NCLR_WAITCONNCLR; break; case PICA_CCLINK_N2NCLE_WAITREP: ccl->state = PICA_CCLINK_N2NCLE_WAITCONNCLE; break; default: ; } } int cclink_allocbufs(struct cclink *ccl) { ccl->buf_p1p2 = malloc(DEFAULT_BUF_SIZE); ccl->buf_p2p1 = malloc(DEFAULT_BUF_SIZE); if (!ccl->buf_p1p2 || !ccl->buf_p2p1) { if (ccl->buf_p1p2) free(ccl->buf_p1p2); if (ccl->buf_p2p1) free(ccl->buf_p2p1); return 0; } ccl->buflen_p1p2 = DEFAULT_BUF_SIZE; ccl->buflen_p2p1 = DEFAULT_BUF_SIZE; return 1; } void cclink_setwaitanontlsshutdown(struct cclink *link, struct newconn *nc) { nc->iconn.cc = link; nc->state = PICA_NEWCONN_WAITANONTLSSHUTDOWN; switch(link->state) { case PICA_CCLINK_LOCAL_WAITCONNCLE: link->sck_p2 = nc->sck; link->state = PICA_CCLINK_LOCAL_WAITCONNCLEANONTLSSHUTDOWN; break; case PICA_CCLINK_LOCAL_WAITCONNCLR: link->sck_p1 = nc->sck; if (!cclink_allocbufs(link)) { cclink_list_delete(link); return; } link->state = PICA_CCLINK_LOCAL_WAITCONNCLRANONTLSSHUTDOWN; break; case PICA_CCLINK_N2NCLR_WAITCONNCLR: link->sck_p1 = nc->sck; if (!cclink_allocbufs(link)) { cclink_list_delete(link); return; } link->state = PICA_CCLINK_N2NCLR_WAITCONNCLRANONTLSSHUTDOWN; break; case PICA_CCLINK_N2NCLE_WAITCONNCLE: link->sck_p2 = nc->sck; if (!cclink_allocbufs(link)) { cclink_list_delete(link); return; } link->state = PICA_CCLINK_N2NCLE_WAITCONNCLEANONTLSSHUTDOWN; break; } } void cclink_activate(struct cclink *ccl) { switch(ccl->state) { case PICA_CCLINK_LOCAL_WAITCONNCLEANONTLSSHUTDOWN: ccl->state = PICA_CCLINK_LOCAL_WAITCONNCLR; break; case PICA_CCLINK_LOCAL_WAITCONNCLRANONTLSSHUTDOWN: ccl->state = PICA_CCLINK_LOCAL_ACTIVE; ccl->ev_read_p1 = event_new(pica_event_base, ccl->sck_p1, EV_READ | EV_PERSIST, process_c2c_read, ccl); ccl->ev_read_p2 = event_new(pica_event_base, ccl->sck_p2, EV_READ | EV_PERSIST, process_c2c_read, ccl); event_add(ccl->ev_read_p1, NULL); event_add(ccl->ev_read_p2, NULL); ccl->ev_write_p1 = event_new(pica_event_base, ccl->sck_p1, EV_WRITE, process_c2c_write, ccl); ccl->ev_write_p2 = event_new(pica_event_base, ccl->sck_p2, EV_WRITE, process_c2c_write, ccl); break; case PICA_CCLINK_N2NCLR_WAITCONNCLRANONTLSSHUTDOWN: ccl->state = PICA_CCLINK_N2NCLR_ACTIVE; ccl->ev_read_p1 = event_new(pica_event_base, ccl->sck_p1, EV_READ | EV_PERSIST, process_c2c_read, ccl); event_add(ccl->ev_read_p1, NULL); ccl->ev_write_p1 = event_new(pica_event_base, ccl->sck_p1, EV_WRITE, process_c2c_write, ccl); break; case PICA_CCLINK_N2NCLE_WAITCONNCLEANONTLSSHUTDOWN: ccl->state = PICA_CCLINK_N2NCLE_ACTIVE; ccl->ev_read_p2 = event_new(pica_event_base, ccl->sck_p2, EV_READ | EV_PERSIST, process_c2c_read, ccl); event_add(ccl->ev_read_p2, NULL); ccl->ev_write_p2 = event_new(pica_event_base, ccl->sck_p2, EV_WRITE, process_c2c_write, ccl); break; } } static void libevent_log_cb(int severity, const char *msg) { switch(severity) { case EVENT_LOG_ERR: PICA_error("libevent error: %s", msg); break; case EVENT_LOG_WARN: case EVENT_LOG_MSG: PICA_info("libevent: %s", msg); break; case EVENT_LOG_DEBUG: PICA_debug1("libevent: %s", msg); break; } } int PICA_node_init() { struct sockaddr_in sd; int ret; int flag; FILE *dh_file; DH *dh = NULL; struct event *listener_event; struct event *timeout_event; struct timeval tv; SSL_load_error_strings(); SSL_library_init(); #ifdef WIN32 WSADATA wsd; WSAStartup(MAKEWORD(2, 2), &wsd); #endif #ifndef WIN32 signal(SIGPIPE, SIG_IGN); #endif event_set_log_callback(libevent_log_cb); pica_event_base = event_base_new(); if (!pica_event_base) PICA_fatal("failed to initialize libevent"); PICA_debug1("libevent method: %s", event_base_get_method(pica_event_base)); //CONF -AF_INET6 ???? listen_comm_sck = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP); if (listen_comm_sck == -1) PICA_fatal("unable to get socket - %s", strerror(errno)); #ifndef WIN32 flag = 1; //enable SO_REUSEADDR if (0 != setsockopt(listen_comm_sck, SOL_SOCKET, SO_REUSEADDR, &flag, sizeof(flag))) { PICA_error("failed to set SO_REUSEADDR on socket: %s", strerror(errno)); } #endif memset(&sd, 0, sizeof(sd)); sd.sin_family = AF_INET; sd.sin_addr.s_addr = INADDR_ANY; //CONF sd.sin_port = (nodecfg.listen_port ? htons(atoi(nodecfg.listen_port)) : htons(PICA_COMM_PORT));//CONF ret = bind(listen_comm_sck, (struct sockaddr*)&sd, sizeof(sd)); if (ret == -1) PICA_fatal("unable to bind socket - %s", strerror(errno)); ret = listen(listen_comm_sck, 20); if (ret == -1) PICA_fatal("unable to listen socket - %s", strerror(errno)); { int ret_comm; unsigned long arg = 1; #ifdef WIN32 ret_comm = ioctlsocket(listen_comm_sck, FIONBIO, &arg); // ret_data=ioctlsocket(listen_data_sck,FIONBIO,&arg); // ret_data=ioctlsocket(listen_node_sck,FIONBIO,&arg); #else ret_comm = ioctl(listen_comm_sck, FIONBIO, (int*)&arg); // ret_data=ioctl(listen_data_sck,FIONBIO,(int*)&arg); // ret_data=ioctl(listen_node_sck,FIONBIO,(int*)&arg); #endif if (ret_comm == -1) PICA_fatal("unable to set non-blocking mode on socket - %s", strerror(errno)); } //#ifdef NO_RAND_DEV //PICA_rand_seed(); //#endif listener_event = event_new(pica_event_base, listen_comm_sck, EV_READ|EV_PERSIST, process_listen_read, NULL); if (!listener_event) PICA_fatal("failed to allocate listener event"); tv.tv_sec = 1; tv.tv_usec = 0; event_add(listener_event, /*&tv*/ NULL); timeout_event = event_new(pica_event_base, -1, EV_TIMEOUT|EV_PERSIST, process_timeout_event, NULL); event_add(timeout_event, &tv); ctx = SSL_CTX_new(TLSv1_2_method()); anon_ctx = SSL_CTX_new(TLSv1_2_method()); if (!ctx || !anon_ctx) PICA_fatal("unable to create SSL_CTX object"); ret = SSL_CTX_set_cipher_list(ctx, PICA_TLS_CIPHERLIST); if (!ret) PICA_fatal("failed to set cipher list " PICA_TLS_CIPHERLIST); SSL_CTX_set_verify(ctx, SSL_VERIFY_PEER, verify_callback); dh_file = fopen(nodecfg.dh_param_file, "r"); if (dh_file) { dh = PEM_read_DHparams(dh_file, NULL, NULL, NULL); fclose(dh_file); } else PICA_fatal("Unable to open DH parameters file %", nodecfg.dh_param_file); ret = SSL_CTX_set_tmp_dh(anon_ctx, dh); if (ret != 1) PICA_fatal("Failed to set DH parameters"); #if OPENSSL_VERSION_NUMBER < 0x10100000L ret = SSL_CTX_set_cipher_list(anon_ctx, PICA_TLS_ANONDHCIPHERLIST); #else ret = SSL_CTX_set_cipher_list(anon_ctx, PICA_TLS_ANONDHCIPHERLIST "@SECLEVEL=0"); #endif if (!ret) PICA_fatal("failed to set cipher list " PICA_TLS_ANONDHCIPHERLIST); newconns_init(); return 1; } void newconn_close(struct newconn* nc) { if (nc->ev) { event_del(nc->ev); event_free(nc->ev); } nc->ev = NULL; CLOSE(nc->sck); PICA_debug3("newconn_close(): nc = %p closed socket %i", nc, nc->sck); nc->sck = -1; nc->pos = 0; if (nc->iconn.cl && nc->type == NEWCONN_C2N) client_list_delete(nc->iconn.cl); if (nc->iconn.cc && nc->type == NEWCONN_C2C) cclink_list_delete(nc->iconn.cc); nc->iconn.cl = NULL; if (nc->anonssl) { SSL_free(nc->anonssl); nc->anonssl = NULL; } nc->type = NEWCONN_UNKNOWN; } void newconns_init() { int i; memset(newconns, 0, sizeof(struct newconn) * MAX_NEWCONNS); for (i = 0; i < MAX_NEWCONNS; i++) { newconns[i].sck = -1; newconns[i].type = NEWCONN_UNKNOWN; } } //функция ищет свободный элемент в массиве структур newconn и возвращает на него указатель //В случае, если свободный элемент не найден, возвращается элемент с наиболее поздней отметкой времени, //при этом соединение закрывается struct newconn* newconn_add(struct newconn *ncs, int *pos) { time_t min; int min_pos = 0; if (*pos < MAX_NEWCONNS && ncs[*pos].sck == -1) { goto ret; } else { *pos = 0; min = ncs->tmst; while(*pos < MAX_NEWCONNS && ncs[*pos].sck != -1) { if (ncs[*pos].tmst < min) { min = ncs[*pos].tmst; min_pos = *pos; } (*pos)++; } if ( (*pos) == MAX_NEWCONNS) { PICA_error("Too many new connections, closing newconn socket %i", ncs[min_pos].sck); newconn_close(&ncs[min_pos]); *pos = min_pos; } } ret: ncs[*pos].sck = -1; ncs[*pos].pos = 0; ncs[*pos].tmst = time(0); return &ncs[(*pos)++]; } void newconn_free(struct newconn* nc) { nc->sck = -1; nc->iconn.cl = NULL; nc->anonssl = NULL; event_del(nc->ev); event_free(nc->ev); nc->ev = NULL; //nc->pos=0; - pos value is being used in processmsgdatastream } int client_tree_add(struct client *ci) { struct client *i; if (client_tree_root) { i = client_tree_root; do { int cmp; cmp = memcmp(ci->id, i->id, PICA_ID_SIZE); if (cmp == 0) return 0; if (cmp < 0) // ci->id < i->id { if (i->left) i = i->left; else { i->left = ci; ci->up = i; return 1; } } else { if (i->right) i = i->right; else { i->right = ci; ci->up = i; return 1; } } } while(1); } else client_tree_root = ci; return 1; } struct client* client_tree_search(const unsigned char *id) { struct client* i_ptr; i_ptr = client_tree_root; //PICA_debug1("client_tree_search: searching for %u...",id); while(i_ptr) { int cmp = memcmp(id, i_ptr->id, PICA_ID_SIZE); if (cmp == 0) return i_ptr; if (cmp < 0) //id < i_ptr->id i_ptr = i_ptr->left; else i_ptr = i_ptr->right; } PICA_debug1("client_tree_search- id not found"); return 0; } void client_tree_print(struct client *c) { char id_buf[2 * PICA_ID_SIZE]; if (!c) { PICA_debug2("client_tree: NULL"); return; } PICA_debug2("client_tree %p: %s state: %u\nUP: %p\nnext_multi: %p\nLEFT: %p, RIGHT: %p\n\n", (void*)c, PICA_id_to_base64(c->id, id_buf), c->state, (void*)c->up, (void*)c->next_multi, (void*)c->left, (void*)c->right); if (c->next_multi) client_tree_print(c->next_multi); if (c->left) client_tree_print(c->left); if (c->right) client_tree_print(c->right); } void client_attach_multi_secondary(struct client *primary, struct client *secondary) { secondary->next_multi = primary->next_multi; primary->next_multi = secondary; } void client_tree_remove(struct client* ci) { struct client** p_link;//указывает на left или right в родительском узле if (!ci->up) { p_link = &client_tree_root; } else { if (ci == ci->up->left) p_link = &(ci->up->left); else p_link = &(ci->up->right); } if (!ci->left && !ci->right) { *p_link = 0; return; } if (ci->left && !ci->right) { ci->left->up = ci->up; *p_link = ci->left; return; } if (!ci->left && ci->right) { ci->right->up = ci->up; *p_link = ci->right; return; } { struct client *lm; lm = ci->right; while(lm->left) lm = lm->left; //поиск самого левого узла правого поддерева if (lm != ci->right) { lm->up->left = lm->right; lm->right = ci->right; } lm->left = ci->left; lm->up = ci->up; *p_link = lm; } } struct client *client_list_addnew(struct newconn *nc) { struct client* ci; ci = (struct client *)calloc(1, sizeof(struct client)); if (!ci) return 0; ci->ssl_comm = SSL_new (ctx); if (!ci->ssl_comm) { free(ci); return 0; } if (client_list_end) { client_list_end->next = ci; ci->prev = client_list_end; client_list_end = ci; } else { client_list_head = ci; client_list_end = ci; } ci->sck_comm = nc->sck; ci->addr = nc->addr; SSL_set_fd(ci->ssl_comm, ci->sck_comm); ci->r_buf = calloc(1, DEFAULT_BUF_SIZE); if (!ci->r_buf) { free(ci); return 0; } ci->buflen_r = DEFAULT_BUF_SIZE; ci->tmst = time(0); ci->ev_read = event_new(pica_event_base, ci->sck_comm, EV_READ | EV_PERSIST, process_c2n_read, ci); ci->ev_write = event_new(pica_event_base, ci->sck_comm, EV_WRITE, process_c2n_write, ci); return ci; } void client_tree_replace_multi(struct client *primary) { struct client** p_link; struct client *secondary; PICA_debug2("client_tree_replace_multi(%p)", (void*)primary); if (!primary->up) { p_link = &client_tree_root; } else { if (primary == primary->up->left) p_link = &(primary->up->left); else p_link = &(primary->up->right); } *p_link = primary->next_multi; secondary = primary->next_multi; secondary->state = primary->state; secondary->up = primary->up; secondary->left = primary->left; secondary->right = primary->right; if (primary->left) primary->left->up = secondary; if (primary->right) primary->right->up = secondary; } void client_list_delete(struct client* ci) { struct client *s; PICA_debug1("client_list_delete(%p)", (void*)ci); SSL_free(ci->ssl_comm); //удаление из списка if (ci->prev) ci->prev->next = ci->next; else client_list_head = ci->next; if (ci->next) ci->next->prev = ci->prev; else client_list_end = ci->prev; //удаление из дерева s = client_tree_search(ci->id); if (s == ci) { if (ci->next_multi) client_tree_replace_multi(ci); else client_tree_remove(ci); client_tree_print(client_tree_root); } else if (s) { while(s) { if (s->next_multi == ci) { s->next_multi = ci->next_multi; break; } s = s->next_multi; } } cclink_list_delete_by_client(ci); if (ci->r_buf) free(ci->r_buf); if (ci->w_buf) free(ci->w_buf); if (ci->ev_read) { event_del(ci->ev_read); event_free(ci->ev_read); } if (ci->ev_write) { event_del(ci->ev_write); event_free(ci->ev_write); } CLOSE(ci->sck_comm); PICA_debug2("closed c2n socket %i", ci->sck_comm); free(ci); } struct PICA_proto_msg* client_wbuf_push (struct client *c, unsigned int msgid, unsigned int size) { if (!c->w_buf) { c->w_buf = calloc(1, DEFAULT_BUF_SIZE + (size / DEFAULT_BUF_SIZE) * DEFAULT_BUF_SIZE ); if (!c->w_buf) { PICA_error("calloc() failed. Out of memory"); return 0; } c->buflen_w = DEFAULT_BUF_SIZE + (size / DEFAULT_BUF_SIZE) * DEFAULT_BUF_SIZE; } if ((c->buflen_w - c->w_pos) < size) { unsigned char *p; p = realloc(c->w_buf, size + c->w_pos); if (p) { c->w_buf = p; c->buflen_w = size + c->w_pos; } else return 0; } event_add(c->ev_write, NULL); struct PICA_proto_msg *mp = (struct PICA_proto_msg *)(c->w_buf + c->w_pos); mp->head[0] = mp->head[1] = msgid; c->w_pos += size; return mp; } int client_rbuf_grow(struct client *c) { unsigned char *p; p = realloc(c->r_buf, c->buflen_r + DEFAULT_BUF_SIZE); if (!p) { PICA_error("realloc() failed"); return 0; } c->r_buf = p; c->buflen_r += DEFAULT_BUF_SIZE; return 1; } struct nodelink *nodelink_list_addnew(struct newconn *nc) { struct nodelink *nl; nl = (struct nodelink*)calloc(1, sizeof(struct nodelink)); if (!nl) return 0; if (nodelink_list_end) { nodelink_list_end->next = nl; nl->prev = nodelink_list_end; nodelink_list_end = nl; } else { nodelink_list_head = nl; nodelink_list_end = nl; } nl->sck = nc->sck; nl->addr = nc->addr; nl->anonssl = nc->anonssl; nc->anonssl = NULL; nl->r_buf = calloc(1, DEFAULT_BUF_SIZE); if (!nl->r_buf) { free(nl); return 0; } nl->buflen_r = DEFAULT_BUF_SIZE; nl->tmst = time(0); nl->ev_read = event_new(pica_event_base, nl->sck, EV_READ | EV_PERSIST, process_n2n_read, nl); nl->ev_write = event_new(pica_event_base, nl->sck, EV_WRITE, process_n2n_write, nl); event_add(nl->ev_read, NULL); nodelink_list_count++; return nl; } void nodelink_list_delete(struct nodelink *n) { PICA_debug1("nodelink_list_delete(%p)", (void*)n); if (n->r_buf) free(n->r_buf); if (n->w_buf) free(n->w_buf); if (n->node_addr) free(n->node_addr); SSL_free(n->anonssl); event_del(n->ev_read); event_free(n->ev_read); event_del(n->ev_write); event_free(n->ev_write); CLOSE(n->sck); PICA_debug2("closed n2n socket %i", n->sck); if (n->prev) n->prev->next = n->next; else nodelink_list_head = n->next; if (n->next) n->next->prev = n->prev; else nodelink_list_end = n->prev; cclink_list_delete_by_nodelink(n); nodelink_list_count--; free(n); } int nodelink_attach_nodeaddr(struct nodelink *nl, unsigned int addr_type, void* nodeaddr, unsigned int nodeaddr_size) { if (nl && nodeaddr) { if (nl->node_addr) free(nl->node_addr); nl->node_addr = malloc(nodeaddr_size); if (!nl->node_addr) return 0;//ERR_CHECK memcpy(nl->node_addr, nodeaddr, nodeaddr_size); nl->addr_type = addr_type; return 1; } return 0; } int nodelink_rbuf_grow(struct nodelink *nl) { unsigned char *p; p = realloc(nl->r_buf, nl->buflen_r + DEFAULT_BUF_SIZE); if (!p) { PICA_error("realloc() failed"); return 0; } nl->r_buf = p; nl->buflen_r += DEFAULT_BUF_SIZE; return 1; } struct PICA_proto_msg* nodelink_wbuf_push (struct nodelink *nl, unsigned int msgid, unsigned int size) { struct PICA_proto_msg *mp; if (!nl->w_buf) { nl->w_buf = calloc(1, DEFAULT_BUF_SIZE ); if (!nl->w_buf) { PICA_error("calloc() failed. Out of memory"); return 0; } nl->buflen_w = DEFAULT_BUF_SIZE; } if ((nl->buflen_w - nl->w_pos) < size) { unsigned char *p; p = realloc(nl->w_buf, size + nl->w_pos); if (p) { nl->w_buf = p; nl->buflen_w = size + nl->w_pos; } else return 0; } event_add(nl->ev_write, NULL); mp = (struct PICA_proto_msg *)(nl->w_buf + nl->w_pos); mp->head[0] = mp->head[1] = msgid; nl->w_pos += size; return mp; } struct nodelink *nodelink_search_by_ipv4addr(in_addr_t ip, in_port_t port) { struct nodelink *nl; nl = nodelink_list_head; while(nl && !( (nl->addr.sin_addr.s_addr == ip && nl->addr.sin_port == port) || (nl->node_addr && nl->addr_type == PICA_PROTO_NEWNODE_IPV4 && ((struct PICA_nodeaddr_ipv4*)nl->node_addr) ->addr == ip && ((struct PICA_nodeaddr_ipv4*)nl->node_addr) ->port == port ) ) ) { nl = nl->next; } return nl; } void process_timeouts_newconn() { clock_t t; int i; t = time(0); for (i = 0; i < MAX_NEWCONNS; i++) { if (newconns[i].sck != -1) { if (newconns[i].tmst < t) { if (t - newconns[i].tmst > NEWCONN_TIMEOUT) { newconn_close(newconns + i); } } else if ( newconns[i].tmst - t > NEWCONN_TIMEOUT) { newconn_close(newconns + i); } } } } void process_timeouts_c2c() { struct cclink *il; il = cclink_list_head; while(il) { switch(il->state) { case PICA_CCLINK_LOCAL_ACTIVE: case PICA_CCLINK_N2NCLR_ACTIVE: case PICA_CCLINK_N2NCLE_ACTIVE: //idle timeout (>=60sec), update timestamp at process_c2c break; default: if ((time(0) - il->tmst) > TMO_CCLINK_WAITACTIVE) { struct cclink *dl; dl = il; il = il->next; cclink_list_delete(dl); continue; } } il = il->next; } } void process_timeouts_c2n() { struct client *i_ptr, *kill_ptr; int ret; i_ptr = client_list_head; kill_ptr = 0; clock_t cur_time = time(0); while(i_ptr) { if (i_ptr->disconnect_ticking) { if (cur_time >= i_ptr->tmst) { if ((cur_time - i_ptr->tmst) > NOREPLY_TIMEOUT) kill_ptr = i_ptr; } else { //2038 year has come, and this code is still working somewhere with 32bit Unix timestamps :) if ((i_ptr->tmst - cur_time ) > NOREPLY_TIMEOUT) kill_ptr = i_ptr; } } else { if (cur_time >= i_ptr->tmst) { if ((cur_time - i_ptr->tmst) > KEEPALIVE_TIMEOUT) { struct PICA_proto_msg *mp; if ((mp = client_wbuf_push(i_ptr, PICA_PROTO_PINGREQ, PICA_PROTO_PINGREQ_SIZE))) { PICA_debug3("sending PINGREQ to %p", i_ptr); RAND_pseudo_bytes(mp->tail, 2); i_ptr->tmst = time(0); i_ptr->disconnect_ticking = 1; } else kill_ptr = i_ptr; } } } i_ptr = i_ptr->next; if (kill_ptr) { PICA_info("Disconnecting user %u due to noreply timeout", kill_ptr -> id); client_list_delete(kill_ptr); kill_ptr = 0; } } } void process_timeouts_n2n() { struct nodelink *i_ptr, *kill_ptr; int ret; i_ptr = nodelink_list_head; kill_ptr = 0; clock_t cur_time = time(0); while(i_ptr) { if (i_ptr->disconnect_ticking) { if (cur_time >= i_ptr->tmst) { if ((cur_time - i_ptr->tmst) > NOREPLY_TIMEOUT) kill_ptr = i_ptr; } else { //2038 year has come, and this code is still working somewhere with 32bit Unix timestamps :) if ((i_ptr->tmst - cur_time ) > NOREPLY_TIMEOUT) kill_ptr = i_ptr; } } else { if (cur_time >= i_ptr->tmst) { if ((cur_time - i_ptr->tmst) > KEEPALIVE_TIMEOUT) { struct PICA_proto_msg *mp; if ((mp = nodelink_wbuf_push(i_ptr, PICA_PROTO_PINGREQ, PICA_PROTO_PINGREQ_SIZE))) { PICA_debug3("sending PINGREQ to node %p", i_ptr); RAND_pseudo_bytes(mp->tail, 2); i_ptr->tmst = time(0); i_ptr->disconnect_ticking = 1; } else kill_ptr = i_ptr; } } } i_ptr = i_ptr->next; if (kill_ptr) { PICA_info("Disconnecting node %p due to noreply timeout", kill_ptr); nodelink_list_delete(kill_ptr); kill_ptr = 0; } } } int process_timeouts() { process_timeouts_newconn(); process_timeouts_c2n(); process_timeouts_c2c(); process_timeouts_n2n(); if (time(0) - skynet_refresh_tmst > SKYNET_REFRESH_TIMEOUT && !nodewait_list) { int clean_treshold = 100 - nodelink_list_count - cclink_list_count; if (clean_treshold <= 0) clean_treshold = 3; PICA_debug3("refreshing nodelist database"); if (nodelink_list_count > 0) { PICA_debug3("trying to clean nodes unreachable after more than %i attempts", clean_treshold); PICA_nodeaddr_cleanold(nodecfg.nodes_db_file, clean_treshold); } PICA_node_joinskynet(nodecfg.nodes_db_file, nodecfg.announced_addr); } return 1; } void process_timeout_event(evutil_socket_t listener, short event, void *arg) { process_timeouts(); process_nodewait(); } void process_listen_read(evutil_socket_t listener, short event, void *arg) { struct newconn *nc; SOCKET s; struct sockaddr_in addr; int addrsize = sizeof(struct sockaddr_in); if (event & EV_READ) { int i = 0; while (i++ < MAX_NEWCONNS / 4 && (s = accept(listen_comm_sck, (struct sockaddr*)&addr, &addrsize)) != -1) { struct event *nc_read, *nc_write; nc = newconn_add(newconns, &newconns_pos); nc->anonssl = SSL_new(anon_ctx); if (!nc->anonssl) { PICA_error("failed to allocate SSL struct"); return; } if (!SSL_set_fd(nc->anonssl, s)) { SSL_free(nc->anonssl); nc->anonssl = NULL; return; } nc->sck = s; nc->addr = addr; nc->state = PICA_NEWCONN_WAITANONTLSACCEPT; IOCTLSETNONBLOCKINGSOCKET(s, 1); PICA_debug1("accepted connection from %.16s, new socket %i", inet_ntoa(addr.sin_addr), s);//IPv6 nc->ev = event_new(pica_event_base, s, EV_READ, process_newconn_read, nc); event_add(nc->ev, NULL); } if (s == -1) { #ifndef WIN32 if (errno == EAGAIN || errno == EWOULDBLOCK || errno == EINTR) return; PICA_error("unable to accept connection: %s", strerror(errno)); #else DWORD err = WSAGetLastError(); if (err != WSAEWOULDBLOCK) PICA_error("unable to accept connection: %u", err); #endif } } //IPv6 } void process_newconn_write(struct newconn *nc) { int ret; switch(nc->state) { case PICA_NEWCONN_SENDINGOK: PICA_debug3("sending OK response to newconn on socket %i...\n", nc->sck); ret = SSL_write(nc->anonssl, msg_INITRESPOK, sizeof msg_INITRESPOK); if (ret < 0 && process_async_ssl_readwrite(nc->anonssl, ret)) break; if (ret <= 0) { newconn_close(nc); break; } if (ret == sizeof msg_INITRESPOK) { if (nc->type == NEWCONN_N2N) { if (!nodelink_list_addnew(nc)) { newconn_close(nc); } newconn_free(nc); } else nc->state = PICA_NEWCONN_WAITANONTLSSHUTDOWN; } break; case PICA_NEWCONN_SENDINGREJECT: PICA_debug3("sending reject response to newconn on socket %i...\n", nc->sck); ret = SSL_write(nc->anonssl, msg_VERDIFFER, sizeof msg_VERDIFFER); if (ret < 0 && process_async_ssl_readwrite(nc->anonssl, ret)) break; if (ret <= 0) { newconn_close(nc); break; } SSL_shutdown(nc->anonssl); SHUTDOWN(nc->sck); newconn_close(nc); } } void process_newconn_read(evutil_socket_t s, short event, void *arg) { int ret; struct newconn *nc = arg; if (event & EV_READ) { switch(nc->state) { case PICA_NEWCONN_WAITANONTLSACCEPT: ret = SSL_accept(nc->anonssl); if (ret < 0 && process_async_ssl_readwrite(nc->anonssl, ret)) break; if (ret <= 0) { newconn_close(nc); break; } else if (ret == 1) { nc->state = PICA_NEWCONN_WAITREQUEST; } case PICA_NEWCONN_WAITREQUEST: ret = SSL_read(nc->anonssl, nc->buf + nc->pos, NEWCONN_BUFSIZE - nc->pos); if (ret < 0 && process_async_ssl_readwrite(nc->anonssl, ret)) break; if (ret <= 0) { newconn_close(nc); break; } nc->pos += ret; if(!PICA_processdatastream(nc->buf, &(nc->pos), nc, _msginfo_newconn, MSGINFO_MSGSNUM(_msginfo_newconn) )) { newconn_close(nc); } break; case PICA_NEWCONN_WAITANONTLSSHUTDOWN: PICA_debug3("waiting for anon TLS shutdown on socket %i...\n", nc->sck); ret = SSL_shutdown(nc->anonssl); if (ret == 0) break; if (ret < 0 && process_async_ssl_readwrite(nc->anonssl, ret)) break; if (ret < 0) { newconn_close(nc); break; } //switch state of client or cclink on successful shutdown if (ret == 1) { SSL_free(nc->anonssl); if (nc->iconn.cl && nc->type == NEWCONN_C2N) { nc->iconn.cl->state = PICA_CLSTATE_TLSNEGOTIATION; event_add(nc->iconn.cl->ev_read, NULL); event_add(nc->iconn.cl->ev_write, NULL); newconn_free(nc); break; } if (nc->iconn.cc && nc->type == NEWCONN_C2C) { cclink_sendcleconnected(nc->iconn.cc); cclink_activate(nc->iconn.cc); newconn_free(nc); break; } } break; } if (nc->sck != -1) process_newconn_write(nc); if (nc->sck != -1) event_add(nc->ev, NULL); } } void process_c2n_read(evutil_socket_t s, short event, void *arg) { struct client *i_ptr = (struct client*)arg; struct client *kill_ptr = 0; int ret; switch(i_ptr->state) { case PICA_CLSTATE_CONNECTED: case PICA_CLSTATE_MULTILOGIN_SECONDARY: ret = SSL_read(i_ptr->ssl_comm, i_ptr->r_buf + i_ptr->r_pos, i_ptr->buflen_r - i_ptr->r_pos); if (!ret) kill_ptr = i_ptr; if (ret < 0) switch(SSL_get_error(i_ptr->ssl_comm, ret)) { case SSL_ERROR_WANT_WRITE: case SSL_ERROR_WANT_READ: break; default: kill_ptr = i_ptr; } if (ret > 0) { const struct PICA_msginfo *msgs = _msginfo_comm; unsigned int nmsgs = MSGINFO_MSGSNUM(_msginfo_comm); if (i_ptr->state == PICA_CLSTATE_MULTILOGIN_SECONDARY) { msgs = _msginfo_multi; nmsgs = MSGINFO_MSGSNUM(_msginfo_multi); } i_ptr->r_pos += ret; if(!PICA_processdatastream(i_ptr->r_buf, &(i_ptr->r_pos), i_ptr, msgs, nmsgs)) kill_ptr = i_ptr; if (i_ptr->buflen_r == i_ptr->r_pos) if (!client_rbuf_grow(i_ptr)) kill_ptr = i_ptr; } break; case PICA_CLSTATE_TLSNEGOTIATION: i_ptr->w_pos = 1; //???? //-------------- <<<<<<<<<<<<<<<<<<<<<<<<---------------------------------------- break; }; if (kill_ptr) { client_list_delete(kill_ptr); kill_ptr = 0; } else if (i_ptr->w_pos) { process_c2n_write(s, event, arg); } } void process_c2c_read(evutil_socket_t s, short event, void *arg) { int ret; struct cclink *cc = (struct cclink*)arg; struct cclink *kill_ptr = 0; if ((cc->state == PICA_CCLINK_LOCAL_ACTIVE || cc->state == PICA_CCLINK_N2NCLR_ACTIVE) && cc->sck_p1 == s && !cc->jam_p1p2 ) { ret = recv(cc->sck_p1, cc->buf_p1p2 + cc->bufpos_p1p2, cc->buflen_p1p2 - cc->bufpos_p1p2, 0); PICA_debug3("process_c2c_read: recv p1p2: ret=%i", ret); if (ret > 0) { cc->bufpos_p1p2 += ret; if (cc->buflen_p1p2 == cc->bufpos_p1p2) { event_del(cc->ev_read_p1); cc->jam_p1p2 = 1; } if (cc->state == PICA_CCLINK_LOCAL_ACTIVE) event_add(cc->ev_write_p2, NULL); else process_c2c_write(-1, event, cc); } else kill_ptr = cc; } if ((cc->state == PICA_CCLINK_LOCAL_ACTIVE || cc->state == PICA_CCLINK_N2NCLE_ACTIVE) && cc->sck_p2 == s && !cc->jam_p2p1 ) { ret = recv(cc->sck_p2, cc->buf_p2p1 + cc->bufpos_p2p1, cc->buflen_p2p1 - cc->bufpos_p2p1, 0); PICA_debug3("process_c2c_read: recv p2p1: ret=%i\n", ret); if (ret > 0) { cc->bufpos_p2p1 += ret; if (cc->buflen_p2p1 == cc->bufpos_p2p1) { event_del(cc->ev_read_p2); cc->jam_p2p1 = 1; } if (cc->state == PICA_CCLINK_LOCAL_ACTIVE) event_add(cc->ev_write_p1, NULL); else process_c2c_write(-1, event, cc); } else kill_ptr = cc; } if (kill_ptr) { cclink_list_delete(kill_ptr); kill_ptr = 0; } } void process_n2n_read(evutil_socket_t s, short event, void *arg) { int ret; struct nodelink *nl = (struct nodelink*)arg; struct nodelink *kill_ptr = 0; ret = SSL_read(nl->anonssl, nl->r_buf + nl->r_pos, nl->buflen_r - nl->r_pos); if (ret > 0) { nl->r_pos += ret; if(!PICA_processdatastream(nl->r_buf, &(nl->r_pos), nl, _msginfo_node, MSGINFO_MSGSNUM(_msginfo_node) )) kill_ptr = nl; if (nl->buflen_r == nl->r_pos) if (!nodelink_rbuf_grow(nl)) kill_ptr = nl; } else { if (ret == 0 || process_async_ssl_readwrite(nl->anonssl, ret) == 0) { PICA_debug3("SSL_read() failed in process_n2n_read()"); kill_ptr = nl; } } if (kill_ptr) { nodelink_list_delete(kill_ptr); kill_ptr = 0; } } void process_c2n_write(evutil_socket_t s, short event, void *arg) { struct client *i_ptr = (struct client*)arg; struct client *kill_ptr = 0; int ret; switch(i_ptr->state) { case PICA_CLSTATE_CONNECTED: case PICA_CLSTATE_MULTILOGIN_SECONDARY: if (!i_ptr->w_pos) break; if (!i_ptr->btw_ssl) i_ptr->btw_ssl = i_ptr->w_pos; ret = SSL_write(i_ptr->ssl_comm, i_ptr->w_buf, i_ptr->btw_ssl); PICA_debug3("process_c2n_write: ret of SSL_write()=%i", ret); if (ret == i_ptr->btw_ssl) { if (ret < i_ptr->w_pos) memmove(i_ptr->w_buf, i_ptr->w_buf + ret, i_ptr->w_pos - ret); //may be it's better to implement ring buffer here i_ptr->w_pos -= i_ptr->btw_ssl; i_ptr->btw_ssl = 0; } if (!ret) kill_ptr = i_ptr; if (ret < 0) { ret = SSL_get_error(i_ptr->ssl_comm, ret); if (ret != SSL_ERROR_WANT_WRITE && ret != SSL_ERROR_WANT_READ) kill_ptr = i_ptr; } break; case PICA_CLSTATE_WAITANONTLSSHUTDOWN: // do nothing here, state will be switched by newconn break; case PICA_CLSTATE_TLSNEGOTIATION: i_ptr->w_pos = 0; ret = SSL_connect(i_ptr->ssl_comm); PICA_debug3("process_c2n_write: SSL_connect()=%i", ret); if (!ret) kill_ptr = i_ptr; if (ret < 0) { int eret; eret = SSL_get_error(i_ptr->ssl_comm, ret); switch(eret) { case SSL_ERROR_WANT_WRITE: i_ptr->w_pos = 1; PICA_debug3("SSL_ERROR_WANT_WRITE"); break; case SSL_ERROR_WANT_READ: PICA_debug3("SSL_ERROR_WANT_READ"); break; default: PICA_debug3("SSL_get_error() = %i", eret); kill_ptr = i_ptr; } } if (ret == 1) //проверить сертификат пользователя { X509* client_cert; struct client *primary; PICA_info("SSL c2n connection established using %s cipher", SSL_get_cipher(i_ptr->ssl_comm)); client_cert = SSL_get_peer_certificate (i_ptr->ssl_comm); if (!client_cert) { //ERR_CHECK //нет сертификата kill_ptr = i_ptr; } ret = PICA_id_from_X509(client_cert, i_ptr->id); if (ret == 0) { PICA_error("Unable to get client id from certificate"); kill_ptr = i_ptr; } X509_free(client_cert); //reject all-zeroes ID { static const char zeroID[PICA_ID_SIZE] = {0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0}; if (memcmp(zeroID, i_ptr->id, PICA_ID_SIZE) == 0) { PICA_error("Rejecting all-zero Pica ID"); kill_ptr = i_ptr; } } if (!kill_ptr && (primary = client_tree_search(i_ptr->id))) { //PICA_info("Disconnecting client %p because other user with same ID = %u is already connected", i_ptr, i_ptr->id); //memset(i_ptr->id, 0, PICA_ID_SIZE);//reset id to all zeros to prevent removing existing ID from previous connection from client_tree //ret = 0;//id already exists in tree i_ptr->state = PICA_CLSTATE_MULTILOGIN_SECONDARY; client_attach_multi_secondary(primary, i_ptr); client_tree_print(client_tree_root); } else { ret = client_tree_add(i_ptr); i_ptr->state = PICA_CLSTATE_CONNECTED; client_tree_print(client_tree_root); } if (!ret) { //ERR_CHECK //не уникальный id kill_ptr = i_ptr; } } break; } if (kill_ptr) { client_list_delete(kill_ptr); kill_ptr = 0; } else if(i_ptr->w_pos) { event_add(i_ptr->ev_write, NULL); } } void process_c2c_write(evutil_socket_t s, short event, void *arg) { int ret; struct cclink *cc = (struct cclink*)arg; struct cclink *kill_ptr = 0; if (cc->bufpos_p1p2) { if (cc->state == PICA_CCLINK_N2NCLR_ACTIVE) //receiver is on remote node { int sendlen; struct PICA_proto_msg *mp; sendlen = (cc->bufpos_p1p2 > PICA_PROTO_N2NMSG_MAXDATASIZE) ? PICA_PROTO_N2NMSG_MAXDATASIZE : cc->bufpos_p1p2; if(mp = nodelink_wbuf_push(cc->callee_node, PICA_PROTO_N2NMSG, sendlen + 4 + 2 * PICA_ID_SIZE)) { *((unsigned short*)mp->tail) = 2 * PICA_ID_SIZE + sendlen; memcpy(mp->tail + 2, cc->caller_id, PICA_ID_SIZE); memcpy(mp->tail + 2 + PICA_ID_SIZE, cc->callee_id, PICA_ID_SIZE); memcpy(mp->tail + 2 + 2 * PICA_ID_SIZE, cc->buf_p1p2, sendlen); ret = sendlen; } else ret = 0; } else if (s == cc->sck_p2) ret = send(cc->sck_p2, cc->buf_p1p2, cc->bufpos_p1p2, MSG_NOSIGNAL); PICA_debug3("process_c2c_write: send p1p2: ret=%i", ret); if (ret > 0) { if (ret < cc->bufpos_p1p2) memmove(cc->buf_p1p2, cc->buf_p1p2 + ret, cc->bufpos_p1p2 - ret); //may be it's better to implement ring buffer here cc->bufpos_p1p2 -= ret; if (cc->bufpos_p1p2 == 0 && cc->sck_p1 != -1) event_add(cc->ev_read_p1, NULL); cc->jam_p1p2 = 0; } if (ret == 0) { kill_ptr = cc; } if (ret == -1) { #ifdef WIN32 ret = WSAGetLastError(); if (ret == WSAEWOULDBLOCK || ret == WSAENOBUFS) cc->jam_p1p2 = 1; else { kill_ptr = cc; } #else ret = errno; if (ret == EAGAIN || ret == ENOBUFS) cc->jam_p1p2 = 1; else { kill_ptr = cc; } #endif } if (!kill_ptr && cc->bufpos_p1p2 && s != -1) event_add(cc->ev_write_p2, NULL); } if (cc->bufpos_p2p1 /*&& (cc->state==PICA_CCLINK_LOCAL_ACTIVE || cc->state==PICA_CCLINK_N2NCLR_ACTIVE)*/) { if (cc->state == PICA_CCLINK_N2NCLE_ACTIVE) //receiver is on remote node { int sendlen; struct PICA_proto_msg *mp; sendlen = (cc->bufpos_p2p1 > PICA_PROTO_N2NMSG_MAXDATASIZE) ? PICA_PROTO_N2NMSG_MAXDATASIZE : cc->bufpos_p2p1; if(mp = nodelink_wbuf_push(cc->caller_node, PICA_PROTO_N2NMSG, sendlen + 4 + 2 * PICA_ID_SIZE)) { *((unsigned short*)mp->tail) = 2 * PICA_ID_SIZE + sendlen; memcpy(mp->tail + 2, cc->callee_id, PICA_ID_SIZE); memcpy(mp->tail + 2 + PICA_ID_SIZE, cc->caller_id, PICA_ID_SIZE); memcpy(mp->tail + 2 + 2 * PICA_ID_SIZE, cc->buf_p2p1, sendlen); ret = sendlen; } else ret = 0; } else if (s == cc->sck_p1) ret = send(cc->sck_p1, cc->buf_p2p1, cc->bufpos_p2p1, MSG_NOSIGNAL); PICA_debug3("process_c2c_write: send p2p1: ret=%i", ret); if (ret > 0) { if (ret < cc->bufpos_p2p1) memmove(cc->buf_p2p1, cc->buf_p2p1 + ret, cc->bufpos_p2p1 - ret); //may be it's better to implement ring buffer here cc->bufpos_p2p1 -= ret; if (cc->bufpos_p2p1 == 0 && cc->sck_p2 != -1) event_add(cc->ev_read_p2, NULL); cc->jam_p2p1 = 0; } if (ret == 0) { kill_ptr = cc; } if (ret == -1) { #ifdef WIN32 ret = WSAGetLastError(); if (ret == WSAEWOULDBLOCK || ret == WSAENOBUFS) cc->jam_p2p1 = 1; else { kill_ptr = cc; } #else ret = errno; if (ret == EAGAIN || ret == ENOBUFS) cc->jam_p2p1 = 1; else { kill_ptr = cc; } #endif } if (!kill_ptr && cc->bufpos_p2p1 && s != -1) event_add(cc->ev_write_p1, NULL); } if (kill_ptr) { cclink_list_delete(kill_ptr); kill_ptr = 0; } } void process_n2n_write(evutil_socket_t s, short event, void *arg) { int ret; struct nodelink *nl = (struct nodelink*)arg; struct nodelink *kill_ptr = 0; if (nl->w_pos > 0) { if (!nl->btw_ssl) nl->btw_ssl = nl->w_pos; ret = SSL_write(nl->anonssl, nl->w_buf, nl->btw_ssl); if (ret == nl->btw_ssl) { if (ret < nl->w_pos) memmove(nl->w_buf, nl->w_buf + ret, nl->w_pos - ret); //may be it's better to implement ring buffer here nl->w_pos -= nl->btw_ssl; nl->btw_ssl = 0; } if (ret == 0 || ret < 0 && process_async_ssl_readwrite(nl->anonssl, ret) == 0) { kill_ptr = nl; } } if (kill_ptr) { nodelink_list_delete(kill_ptr); kill_ptr = 0; } else if (nl->w_pos > 0) { event_add(nl->ev_write, NULL); } } void process_nodewait() { struct nodewait *kill_ptr = 0, *nw = nodewait_list; while(nw) { //PICA_debug3("checking nodewait pointer %p", nw); switch(nw->state) { case PICA_NODEWAIT_RESOLVED: { int skip = 0; struct nodelink *nl; nl = nodelink_list_head; while(nl && !skip)//check existing connections { struct addrinfo *p_ai = nw->ai; while(p_ai) //IPv6 { struct sockaddr_in in = *(struct sockaddr_in*)p_ai->ai_addr; if (nl->addr.sin_family == in.sin_family && nl->addr.sin_addr.s_addr == in.sin_addr.s_addr && nl->addr.sin_port == in.sin_port) { skip = 1; break; } p_ai = p_ai->ai_next; } nl = nl->next; } if (!skip) nodewait_start_connect(nw); else { PICA_debug3("skipping node %.255s %u , connection already exist", nw->addr.addr, nw->addr.port); kill_ptr = nw; } } break; case PICA_NODEWAIT_CONNECTED: { struct nodelink *nlp; PICA_debug1("connected to %.255s %u, sending request ", nw->addr.addr, nw->addr.port); PICA_debug2("connection socket is %i", nw->nc.sck); nlp = nodelink_list_addnew(&nw->nc); if (nlp) { struct PICA_proto_msg *mp; mp = nodelink_wbuf_push(nlp, PICA_PROTO_NODECONNREQ, PICA_PROTO_NODECONNREQ_SIZE); if (mp) { mp->tail[0] = PICA_PROTO_VER_HIGH; mp->tail[1] = PICA_PROTO_VER_LOW; } } PICA_nodeaddr_update(nodecfg.nodes_db_file, &nw->addr, 1); kill_ptr = nw; } break; case PICA_NODEWAIT_RESOLVING_FAILED: { PICA_debug1("resolving %.255s %u failed: %s", nw->addr.addr, nw->addr.port, gai_strerror(nw->ai_errorcode)); PICA_nodeaddr_update(nodecfg.nodes_db_file, &nw->addr, 0); kill_ptr = nw; } break; case PICA_NODEWAIT_CONNECT_FAILED: { PICA_debug1("connection to %.255s %u failed", nw->addr.addr, nw->addr.port); PICA_debug2("socket %i is closed", nw->nc.sck); PICA_nodeaddr_update(nodecfg.nodes_db_file, &nw->addr, 0); kill_ptr = nw; } break; } nw = nw->next; if (kill_ptr) { nodewait_list_delete(kill_ptr); kill_ptr = 0; } } } int node_loop() { int ret; do { ret = event_base_loop(pica_event_base, EVLOOP_ONCE); } while(stop_node_loop == 0);////<<<<<<<<<< return ret; } void print_usage() { puts("Usage: pica-node [-f config-file]\n\ \n\ Options:\n\ -f configuration file\n\ -h show this text\n\ -v verbose\n\ -V print version info\n"); } void print_version() { puts("Pica Pica Node v"PACKAGE_VERSION" \nCopyright (c) 2012 - 2018 Anton Sviridenko\n"); puts("protocol version "PICA_PROTO_VER_STRING"\n"); puts("License GPLv3+: GNU GPL version 3 or later \n\ This is free software: you are free to change and redistribute it.\n\ There is NO WARRANTY, to the extent permitted by law.\n"); ////----include this only if OpenSSL is statically linked puts("This product includes software developed by the OpenSSL Project\n\ for use in the OpenSSL Toolkit (http://www.openssl.org/)\n"); puts("This product includes cryptographic software written by\n\ Eric Young (eay@cryptsoft.com)\n"); puts(SSLeay_version(SSLEAY_VERSION)); ////--- puts("\nThis product includes iniParser\n\ Copyright (c) 2000-2012 by Nicolas Devillard.\n\ MIT License"); } void process_cmdline(int argc, char** argv) { int opt; int verbosity = 0; while((opt = getopt(argc, argv, "f:hvV")) != -1) { switch(opt) { case 'f': nodecfg.config_file = strdup(optarg); break; case 'h': print_usage(); exit(0); break; case 'v': verbosity++; break; case 'V': print_version(); exit(0); break; default: PICA_error("unknown option - %c", optopt); print_usage(); exit(-1); } } if (verbosity > 3) verbosity = 3; PICA_set_loglevel(PICA_LOG_INFO + verbosity); } void PICA_node_joinskynet(const char* addrlistfilename, const char *my_addr) { struct PICA_nodeaddr *nap, *addrlist_h = 0; struct nodelink *nl; int ret; skynet_refresh_tmst = time(0); ret = PICA_nodeaddr_list_load(addrlistfilename, &addrlist_h);//MEM if (ret <= 0) return; nap = addrlist_h; while(nap) { int skip = 0; if (0 == strncmp(nap->addr, my_addr, 256) && nap->port == atoi(nodecfg.listen_port)) { PICA_debug3("Skipping self address %.255s port %u", nap->addr, nap->port); skip = 1; } if (nodecfg.disable_reserved_addrs && INADDR_NONE != inet_addr(nap->addr) && PICA_is_reserved_addr_ipv4(inet_addr(nap->addr))) { PICA_debug3("Skipping IP from private range %.255s port %u", nap->addr, nap->port); skip = 1; } nl = nodelink_list_head; //check if connection to node with current address is already established while(nl && !skip) { if (nl->node_addr) { switch(nl->addr_type) { case PICA_PROTO_NEWNODE_IPV4: if (0 == strncmp(nap->addr, inet_ntoa(*(struct in_addr*) & (((struct PICA_nodeaddr_ipv4*)nl->node_addr)->addr)), 16) && nap->port == ntohs(((struct PICA_nodeaddr_ipv4*)nl->node_addr)->port)) skip = 1; break; //IPv6 } } if (0 == strncmp(nap->addr, inet_ntoa(nl->addr.sin_addr), 16) && nap->port == ntohs(nl->addr.sin_port)) skip = 1; nl = nl->next; } if (!skip) nodewait_start_resolve(nap); nap = nap->next; } PICA_nodeaddr_list_free(addrlist_h); } #ifdef WIN32 static int is_service = 0; static SERVICE_STATUS_HANDLE service_status_handle; static SERVICE_STATUS service_status; static int service_argc; static char **service_argv; void ReportSvcStatus( DWORD dwCurrentState, DWORD dwWin32ExitCode, DWORD dwWaitHint) { static DWORD dwCheckPoint = 1; // Fill in the SERVICE_STATUS structure. service_status.dwCurrentState = dwCurrentState; service_status.dwWin32ExitCode = dwWin32ExitCode; service_status.dwWaitHint = dwWaitHint; if (dwCurrentState == SERVICE_START_PENDING) service_status.dwControlsAccepted = 0; else service_status.dwControlsAccepted = SERVICE_ACCEPT_STOP; if ( (dwCurrentState == SERVICE_RUNNING) || (dwCurrentState == SERVICE_STOPPED) ) service_status.dwCheckPoint = 0; else service_status.dwCheckPoint = dwCheckPoint++; // Report the status of the service to the SCM. SetServiceStatus( service_status_handle, &service_status); } #endif int pica_node_main(int argc, char** argv) { uint32_t pid; process_cmdline(argc, argv); nodecfg.config_file = (nodecfg.config_file ? nodecfg.config_file : PICA_NODECONFIG_DEF_CONFIG_FILE); #ifdef WIN32 pid = GetCurrentProcessId(); #else pid = getpid(); #endif PICA_info("pica-node v%s started. PID: %u, protocol version: %s", PACKAGE_VERSION, pid, PICA_PROTO_VER_STRING); if (!PICA_nodeconfig_load(nodecfg.config_file)) { PICA_fatal("Unable to load configuration. Please check if config file %s exists and has correct permissions", nodecfg.config_file); } PICA_debug1("nodecfg.config_file = %s", nodecfg.config_file); PICA_debug1("nodecfg.announced_addr = %s", nodecfg.announced_addr); PICA_debug1("nodecfg.listen_port = %s", nodecfg.listen_port); PICA_debug1("nodecfg.nodes_db_file = %s", nodecfg.nodes_db_file); PICA_debug1("nodecfg.disable_reserved_addrs = %s", nodecfg.disable_reserved_addrs ? "yes" : "no"); if (!PICA_node_init()) return -1; #ifdef WIN32 if (is_service) { ReportSvcStatus(SERVICE_RUNNING, NO_ERROR, 0); PICA_info("pica-node started as a Windows service"); } else { PICA_info("pica-node started as a normal Windows process."); } #endif if (strcmp("autoconfigure", nodecfg.announced_addr) == 0) { in_addr_t guess; struct in_addr in; guess = PICA_guess_listening_addr_ipv4(); in.s_addr = guess; free(nodecfg.announced_addr); nodecfg.announced_addr = strdup(inet_ntoa(in)); PICA_debug1("guessed interface address to be announced: %s", nodecfg.announced_addr); #ifdef HAVE_LIBMINIUPNPC if (PICA_is_reserved_addr_ipv4(guess) && nodecfg.disable_reserved_addrs) { char public_ip[64]; int ret; PICA_info("trying to get global IP with UPnP..."); ret = PICA_upnp_autoconfigure_ipv4(atoi(nodecfg.listen_port), atoi(nodecfg.listen_port), public_ip); if (ret) { free(nodecfg.announced_addr); nodecfg.announced_addr = strdup(public_ip); } } #endif PICA_info("autoconfigured announced address %s port %s", nodecfg.announced_addr, nodecfg.listen_port); } if (INADDR_NONE == inet_addr(nodecfg.announced_addr) || INADDR_ANY == inet_addr(nodecfg.announced_addr)) { PICA_fatal("announced_addr (%.16s) is invalid or not configured. Please set correct public IP address of your pica-node instance in config file.", nodecfg.announced_addr ); } if (nodecfg.disable_reserved_addrs && PICA_is_reserved_addr_ipv4(inet_addr(nodecfg.announced_addr))) { PICA_fatal("announced_addr (%.16s) is in private range and is unacceptable for global Internet. Private ranges are disabled by node configuration.", nodecfg.announced_addr ); } PICA_node_joinskynet(nodecfg.nodes_db_file, nodecfg.announced_addr);//CONF-CONF имя файла с адресами узлов, свой адрес my_addr = nodecfg.announced_addr;//TEMP FIXME return node_loop(); } #ifdef WIN32 void WINAPI SvcHandler(DWORD ctl) { switch(ctl) { case SERVICE_CONTROL_STOP: ReportSvcStatus(SERVICE_STOP_PENDING, NO_ERROR, 0); stop_node_loop = 1; ReportSvcStatus(service_status.dwCurrentState, NO_ERROR, 0); return; case SERVICE_CONTROL_INTERROGATE: break; default: break; } } void WINAPI winservice_main(DWORD argc, LPTSTR *argv) { char filepath[MAX_PATH]; int ret; HANDLE logfile; service_status_handle = RegisterServiceCtrlHandler("pica-node service", SvcHandler); if (!service_status_handle) return; service_status.dwServiceType = SERVICE_WIN32_OWN_PROCESS; service_status.dwServiceSpecificExitCode = 0; if (S_OK != SHGetFolderPath(NULL, CSIDL_COMMON_APPDATA, NULL, SHGFP_TYPE_CURRENT, filepath)) return; strncat_s(filepath, MAX_PATH, "\\pica-node.log", MAX_PATH); freopen(filepath, "w", stdout); freopen(filepath, "w", stderr); //set current directory to executable location directory GetModuleFileName(GetModuleHandle(NULL), filepath, MAX_PATH); //strip exe filename part *(strrchr(filepath, '\\') + 1) = 0; SetCurrentDirectory(filepath); ReportSvcStatus(SERVICE_START_PENDING, NO_ERROR, 3000); is_service = 1; ret = pica_node_main(service_argc, service_argv); ReportSvcStatus(SERVICE_STOPPED, ret, 0); } #endif int main(int argc, char** argv) { #ifndef WIN32 return pica_node_main(argc, argv); #else SERVICE_TABLE_ENTRY DispatchTable[] = { {"pica-node service", winservice_main}, {NULL, NULL} }; service_argc = argc; service_argv = argv; //if process is started as a service, //function will return only after service is stopped if (StartServiceCtrlDispatcher(DispatchTable)) return 0; else if (GetLastError() == ERROR_FAILED_SERVICE_CONTROLLER_CONNECT)// not a service return pica_node_main(argc, argv); //failed to start service return -1; #endif }