server.c 74 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011101210131014101510161017101810191020102110221023102410251026102710281029103010311032103310341035103610371038103910401041104210431044104510461047104810491050105110521053105410551056105710581059106010611062106310641065106610671068106910701071107210731074107510761077107810791080108110821083108410851086108710881089109010911092109310941095109610971098109911001101110211031104110511061107110811091110111111121113111411151116111711181119112011211122112311241125112611271128112911301131113211331134113511361137113811391140114111421143114411451146114711481149115011511152115311541155115611571158115911601161116211631164116511661167116811691170117111721173117411751176117711781179118011811182118311841185118611871188118911901191119211931194119511961197119811991200120112021203120412051206120712081209121012111212121312141215121612171218121912201221122212231224122512261227122812291230123112321233123412351236123712381239124012411242124312441245124612471248124912501251125212531254125512561257125812591260126112621263126412651266126712681269127012711272127312741275127612771278127912801281128212831284128512861287128812891290129112921293129412951296129712981299130013011302130313041305130613071308130913101311131213131314131513161317131813191320132113221323132413251326132713281329133013311332133313341335133613371338133913401341134213431344134513461347134813491350135113521353135413551356135713581359136013611362136313641365136613671368136913701371137213731374137513761377137813791380138113821383138413851386138713881389139013911392139313941395139613971398139914001401140214031404140514061407140814091410141114121413141414151416141714181419142014211422142314241425142614271428142914301431143214331434143514361437143814391440144114421443144414451446144714481449145014511452145314541455145614571458145914601461146214631464146514661467146814691470147114721473147414751476147714781479148014811482148314841485148614871488148914901491149214931494149514961497149814991500150115021503150415051506150715081509151015111512151315141515151615171518151915201521152215231524152515261527152815291530153115321533153415351536153715381539154015411542154315441545154615471548154915501551155215531554155515561557155815591560156115621563156415651566156715681569157015711572157315741575157615771578157915801581158215831584158515861587158815891590159115921593159415951596159715981599160016011602160316041605160616071608160916101611161216131614161516161617161816191620162116221623162416251626162716281629163016311632163316341635163616371638163916401641164216431644164516461647164816491650165116521653165416551656165716581659166016611662166316641665166616671668166916701671167216731674167516761677167816791680168116821683168416851686168716881689169016911692169316941695169616971698169917001701170217031704170517061707170817091710171117121713171417151716171717181719172017211722172317241725172617271728172917301731173217331734173517361737173817391740174117421743174417451746174717481749175017511752175317541755175617571758175917601761176217631764176517661767176817691770177117721773177417751776177717781779178017811782178317841785178617871788178917901791179217931794179517961797179817991800180118021803180418051806180718081809181018111812181318141815181618171818181918201821182218231824182518261827182818291830183118321833183418351836183718381839184018411842184318441845184618471848184918501851185218531854185518561857185818591860186118621863186418651866186718681869187018711872187318741875187618771878187918801881188218831884188518861887188818891890189118921893189418951896189718981899190019011902190319041905190619071908190919101911191219131914191519161917191819191920192119221923192419251926192719281929193019311932193319341935193619371938193919401941194219431944194519461947194819491950195119521953195419551956195719581959196019611962196319641965196619671968196919701971197219731974197519761977197819791980198119821983198419851986198719881989199019911992199319941995199619971998199920002001200220032004200520062007200820092010201120122013201420152016201720182019202020212022202320242025202620272028202920302031203220332034203520362037203820392040204120422043204420452046204720482049205020512052205320542055205620572058205920602061206220632064206520662067206820692070207120722073207420752076207720782079208020812082208320842085208620872088208920902091209220932094209520962097209820992100210121022103210421052106210721082109211021112112211321142115211621172118211921202121212221232124212521262127212821292130213121322133213421352136213721382139214021412142214321442145214621472148214921502151215221532154215521562157215821592160216121622163216421652166216721682169217021712172217321742175217621772178217921802181218221832184218521862187218821892190219121922193219421952196219721982199220022012202220322042205220622072208220922102211221222132214221522162217221822192220222122222223222422252226222722282229223022312232223322342235223622372238223922402241224222432244224522462247224822492250225122522253225422552256225722582259226022612262226322642265226622672268226922702271227222732274227522762277227822792280228122822283228422852286228722882289229022912292229322942295229622972298229923002301230223032304230523062307230823092310231123122313231423152316231723182319232023212322232323242325232623272328232923302331233223332334233523362337233823392340234123422343234423452346234723482349235023512352235323542355235623572358235923602361236223632364236523662367236823692370237123722373237423752376237723782379238023812382238323842385238623872388238923902391239223932394
  1. /**
  2. * @file server.c
  3. * @author Ambroz Bizjak <ambrop7@gmail.com>
  4. *
  5. * @section LICENSE
  6. *
  7. * Redistribution and use in source and binary forms, with or without
  8. * modification, are permitted provided that the following conditions are met:
  9. * 1. Redistributions of source code must retain the above copyright
  10. * notice, this list of conditions and the following disclaimer.
  11. * 2. Redistributions in binary form must reproduce the above copyright
  12. * notice, this list of conditions and the following disclaimer in the
  13. * documentation and/or other materials provided with the distribution.
  14. * 3. Neither the name of the author nor the
  15. * names of its contributors may be used to endorse or promote products
  16. * derived from this software without specific prior written permission.
  17. *
  18. * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND
  19. * ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
  20. * WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
  21. * DISCLAIMED. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR ANY
  22. * DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES
  23. * (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
  24. * LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND
  25. * ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
  26. * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
  27. * SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
  28. */
  29. #include <stdint.h>
  30. #include <stdlib.h>
  31. #include <string.h>
  32. #include <stddef.h>
  33. #include <stdarg.h>
  34. // NSPR and NSS
  35. #include <prinit.h>
  36. #include <prio.h>
  37. #include <prerror.h>
  38. #include <prtypes.h>
  39. #include <nss/nss.h>
  40. #include <nss/ssl.h>
  41. #include <nss/cert.h>
  42. #include <nss/keyhi.h>
  43. #include <nss/secasn1.h>
  44. // BadVPN
  45. #include <misc/version.h>
  46. #include <misc/debug.h>
  47. #include <misc/offset.h>
  48. #include <misc/nsskey.h>
  49. #include <misc/byteorder.h>
  50. #include <misc/loglevel.h>
  51. #include <misc/loggers_string.h>
  52. #include <misc/open_standard_streams.h>
  53. #include <misc/compare.h>
  54. #include <misc/bsize.h>
  55. #include <predicate/BPredicate.h>
  56. #include <base/DebugObject.h>
  57. #include <base/BLog.h>
  58. #include <system/BSignal.h>
  59. #include <system/BTime.h>
  60. #include <system/BNetwork.h>
  61. #include <security/BRandom.h>
  62. #include <nspr_support/DummyPRFileDesc.h>
  63. #include <threadwork/BThreadWork.h>
  64. #ifndef BADVPN_USE_WINAPI
  65. #include <base/BLog_syslog.h>
  66. #endif
  67. #include <server/server.h>
  68. #include <generated/blog_channel_server.h>
  69. #define LOGGER_STDOUT 1
  70. #define LOGGER_SYSLOG 2
  71. // parsed command-line options
  72. struct {
  73. int help;
  74. int version;
  75. int logger;
  76. #ifndef BADVPN_USE_WINAPI
  77. char *logger_syslog_facility;
  78. char *logger_syslog_ident;
  79. #endif
  80. int loglevel;
  81. int loglevels[BLOG_NUM_CHANNELS];
  82. int threads;
  83. int use_threads_for_ssl_handshake;
  84. int use_threads_for_ssl_data;
  85. int ssl;
  86. char *nssdb;
  87. char *server_cert_name;
  88. char *listen_addrs[MAX_LISTEN_ADDRS];
  89. int num_listen_addrs;
  90. char *comm_predicate;
  91. char *relay_predicate;
  92. int client_socket_sndbuf;
  93. int max_clients;
  94. } options;
  95. // listen addresses
  96. BAddr listen_addrs[MAX_LISTEN_ADDRS];
  97. int num_listen_addrs;
  98. // communication predicate
  99. BPredicate comm_predicate;
  100. // communication predicate functions
  101. BPredicateFunction comm_predicate_func_p1name;
  102. BPredicateFunction comm_predicate_func_p2name;
  103. BPredicateFunction comm_predicate_func_p1addr;
  104. BPredicateFunction comm_predicate_func_p2addr;
  105. // variables when evaluating the predicate, adjusted before every evaluation
  106. const char *comm_predicate_p1name;
  107. const char *comm_predicate_p2name;
  108. BIPAddr comm_predicate_p1addr;
  109. BIPAddr comm_predicate_p2addr;
  110. // relay predicate
  111. BPredicate relay_predicate;
  112. // gateway predicate functions
  113. BPredicateFunction relay_predicate_func_pname;
  114. BPredicateFunction relay_predicate_func_rname;
  115. BPredicateFunction relay_predicate_func_paddr;
  116. BPredicateFunction relay_predicate_func_raddr;
  117. // variables when evaluating the comm_predicate, adjusted before every evaluation
  118. const char *relay_predicate_pname;
  119. const char *relay_predicate_rname;
  120. BIPAddr relay_predicate_paddr;
  121. BIPAddr relay_predicate_raddr;
  122. // i/o system
  123. BReactor ss;
  124. // thread work dispatcher
  125. BThreadWorkDispatcher twd;
  126. // server certificate if using SSL
  127. CERTCertificate *server_cert;
  128. // server private key if using SSL
  129. SECKEYPrivateKey *server_key;
  130. // model NSPR file descriptor to speed up client initialization
  131. PRFileDesc model_dprfd;
  132. PRFileDesc *model_prfd;
  133. // listeners
  134. BListener listeners[MAX_LISTEN_ADDRS];
  135. int num_listeners;
  136. // number of connected clients
  137. int clients_num;
  138. // ID assigned to last connected client
  139. peerid_t clients_nextid;
  140. // clients list
  141. LinkedList1 clients;
  142. // clients tree (by ID)
  143. BAVL clients_tree;
  144. // prints help text to standard output
  145. static void print_help (const char *name);
  146. // prints program name and version to standard output
  147. static void print_version (void);
  148. // parses the command line
  149. static int parse_arguments (int argc, char *argv[]);
  150. // processes certain command line options
  151. static int process_arguments (void);
  152. static int ssl_flags (void);
  153. // handler for program termination request
  154. static void signal_handler (void *unused);
  155. // listener handler, accepts new clients
  156. static void listener_handler (BListener *listener);
  157. // frees resources used by a client
  158. static void client_dealloc (struct client_data *client);
  159. static int client_compute_buffer_size (struct client_data *client);
  160. // initializes the I/O porition of the client
  161. static int client_init_io (struct client_data *client);
  162. // deallocates the I/O portion of the client. Must have no outgoing flows.
  163. static void client_dealloc_io (struct client_data *client);
  164. // removes a client
  165. static void client_remove (struct client_data *client);
  166. // job to finish removal after clients are informed
  167. static void client_dying_job (struct client_data *client);
  168. // appends client log prefix
  169. static void client_logfunc (struct client_data *client);
  170. // passes a message to the logger, prepending about the client
  171. static void client_log (struct client_data *client, int level, const char *fmt, ...);
  172. // client activity timer handler. Removes the client.
  173. static void client_disconnect_timer_handler (struct client_data *client);
  174. // BConnection handler
  175. static void client_connection_handler (struct client_data *client, int event);
  176. // BSSLConnection handler
  177. static void client_sslcon_handler (struct client_data *client, int event);
  178. // decoder handler
  179. static void client_decoder_handler_error (struct client_data *client);
  180. // provides a buffer for sending a control packet to the client
  181. static int client_start_control_packet (struct client_data *client, void **data, int len);
  182. // submits a packet written after client_start_control_packet
  183. static void client_end_control_packet (struct client_data *client, uint8_t id);
  184. // sends a newclient message to a client
  185. static int client_send_newclient (struct client_data *client, struct client_data *nc, int relay_server, int relay_client);
  186. // sends an endclient message to a client
  187. static int client_send_endclient (struct client_data *client, peerid_t end_id);
  188. // handler for packets received from the client
  189. static void client_input_handler_send (struct client_data *client, uint8_t *data, int data_len);
  190. // processes hello packets from clients
  191. static void process_packet_hello (struct client_data *client, uint8_t *data, int data_len);
  192. // processes outmsg packets from clients
  193. static void process_packet_outmsg (struct client_data *client, uint8_t *data, int data_len);
  194. // processes resetpeer packets from clients
  195. static void process_packet_resetpeer (struct client_data *client, uint8_t *data, int data_len);
  196. // processes acceptpeer packets from clients
  197. static void process_packet_acceptpeer (struct client_data *client, uint8_t *data, int data_len);
  198. // creates a peer flow
  199. static struct peer_flow * peer_flow_create (struct client_data *src_client, struct client_data *dest_client);
  200. // deallocates a peer flow
  201. static void peer_flow_dealloc (struct peer_flow *flow);
  202. static int peer_flow_init_io (struct peer_flow *flow);
  203. static void peer_flow_free_io (struct peer_flow *flow);
  204. // disconnects the source client from a peer flow
  205. static void peer_flow_disconnect (struct peer_flow *flow);
  206. // provides a buffer for sending a peer-to-peer packet
  207. static int peer_flow_start_packet (struct peer_flow *flow, void **data, int len);
  208. // submits a peer-to-peer packet written after peer_flow_start_packet
  209. static void peer_flow_end_packet (struct peer_flow *flow, uint8_t type);
  210. // handler called by the queue when a peer flow can be freed after its source has gone away
  211. static void peer_flow_handler_canremove (struct peer_flow *flow);
  212. static void peer_flow_start_reset (struct peer_flow *flow);
  213. static void peer_flow_drive_reset (struct peer_flow *flow);
  214. static void peer_flow_reset_qflow_handler_busy (struct peer_flow *flow);
  215. // resets clients knowledge after the timer expires
  216. static void peer_flow_reset_timer_handler (struct peer_flow *flow);
  217. // generates a client ID to be used for a newly connected client
  218. static peerid_t new_client_id (void);
  219. // finds a client by its ID
  220. static struct client_data * find_client_by_id (peerid_t id);
  221. // checks if two clients are allowed to communicate. May depend on the order
  222. // of the clients.
  223. static int clients_allowed (struct client_data *client1, struct client_data *client2);
  224. // communication predicate function p1name
  225. static int comm_predicate_func_p1name_cb (void *user, void **args);
  226. // communication predicate function p2name
  227. static int comm_predicate_func_p2name_cb (void *user, void **args);
  228. // communication predicate function p1addr
  229. static int comm_predicate_func_p1addr_cb (void *user, void **args);
  230. // communication predicate function p2addr
  231. static int comm_predicate_func_p2addr_cb (void *user, void **args);
  232. // checks if relay is allowed for a client through another client
  233. static int relay_allowed (struct client_data *client, struct client_data *relay);
  234. // relay predicate function pname
  235. static int relay_predicate_func_pname_cb (void *user, void **args);
  236. // relay predicate function rname
  237. static int relay_predicate_func_rname_cb (void *user, void **args);
  238. // relay predicate function paddr
  239. static int relay_predicate_func_paddr_cb (void *user, void **args);
  240. // relay predicate function raddr
  241. static int relay_predicate_func_raddr_cb (void *user, void **args);
  242. // comparator for peerid_t used in AVL tree
  243. static int peerid_comparator (void *unused, peerid_t *p1, peerid_t *p2);
  244. static struct peer_know * create_know (struct client_data *from, struct client_data *to, int relay_server, int relay_client);
  245. static void remove_know (struct peer_know *k);
  246. static void know_inform_job_handler (struct peer_know *k);
  247. static void uninform_know (struct peer_know *k);
  248. static void know_uninform_job_handler (struct peer_know *k);
  249. static int launch_pair (struct peer_flow *flow_to);
  250. // find flow from a client to some client
  251. static struct peer_flow * find_flow (struct client_data *client, peerid_t dest_id);
  252. int main (int argc, char *argv[])
  253. {
  254. if (argc <= 0) {
  255. return 1;
  256. }
  257. // open standard streams
  258. open_standard_streams();
  259. // parse command-line arguments
  260. if (!parse_arguments(argc, argv)) {
  261. fprintf(stderr, "Failed to parse arguments\n");
  262. print_help(argv[0]);
  263. goto fail0;
  264. }
  265. // handle --help and --version
  266. if (options.help) {
  267. print_version();
  268. print_help(argv[0]);
  269. return 0;
  270. }
  271. if (options.version) {
  272. print_version();
  273. return 0;
  274. }
  275. // initialize logger
  276. switch (options.logger) {
  277. case LOGGER_STDOUT:
  278. BLog_InitStdout();
  279. break;
  280. #ifndef BADVPN_USE_WINAPI
  281. case LOGGER_SYSLOG:
  282. if (!BLog_InitSyslog(options.logger_syslog_ident, options.logger_syslog_facility)) {
  283. fprintf(stderr, "Failed to initialize syslog logger\n");
  284. goto fail0;
  285. }
  286. break;
  287. #endif
  288. default:
  289. ASSERT(0);
  290. }
  291. // configure logger channels
  292. for (int i = 0; i < BLOG_NUM_CHANNELS; i++) {
  293. if (options.loglevels[i] >= 0) {
  294. BLog_SetChannelLoglevel(i, options.loglevels[i]);
  295. }
  296. else if (options.loglevel >= 0) {
  297. BLog_SetChannelLoglevel(i, options.loglevel);
  298. }
  299. }
  300. BLog(BLOG_NOTICE, "initializing "GLOBAL_PRODUCT_NAME" "PROGRAM_NAME" "GLOBAL_VERSION);
  301. if (options.ssl) {
  302. // initialize NSPR
  303. PR_Init(PR_USER_THREAD, PR_PRIORITY_NORMAL, 0);
  304. // initialize i/o layer types
  305. if (!DummyPRFileDesc_GlobalInit()) {
  306. BLog(BLOG_ERROR, "DummyPRFileDesc_GlobalInit failed");
  307. goto fail01;
  308. }
  309. if (!BSSLConnection_GlobalInit()) {
  310. BLog(BLOG_ERROR, "BSSLConnection_GlobalInit failed");
  311. goto fail01;
  312. }
  313. // initialize NSS
  314. if (NSS_Init(options.nssdb) != SECSuccess) {
  315. BLog(BLOG_ERROR, "NSS_Init failed (%d)", (int)PR_GetError());
  316. goto fail01;
  317. }
  318. if (NSS_SetDomesticPolicy() != SECSuccess) {
  319. BLog(BLOG_ERROR, "NSS_SetDomesticPolicy failed (%d)", (int)PR_GetError());
  320. goto fail02;
  321. }
  322. // initialize server cache
  323. if (SSL_ConfigServerSessionIDCache(0, 0, 0, NULL) != SECSuccess) {
  324. BLog(BLOG_ERROR, "SSL_ConfigServerSessionIDCache failed (%d)", (int)PR_GetError());
  325. goto fail02;
  326. }
  327. // open server certificate and private key
  328. if (!open_nss_cert_and_key(options.server_cert_name, &server_cert, &server_key)) {
  329. BLog(BLOG_ERROR, "Cannot open certificate and key");
  330. goto fail03;
  331. }
  332. // initialize model SSL fd
  333. DummyPRFileDesc_Create(&model_dprfd);
  334. if (!(model_prfd = SSL_ImportFD(NULL, &model_dprfd))) {
  335. BLog(BLOG_ERROR, "SSL_ImportFD failed");
  336. ASSERT_FORCE(PR_Close(&model_dprfd) == PR_SUCCESS)
  337. goto fail04;
  338. }
  339. // set server certificate
  340. if (SSL_ConfigSecureServer(model_prfd, server_cert, server_key, NSS_FindCertKEAType(server_cert)) != SECSuccess) {
  341. BLog(BLOG_ERROR, "SSL_ConfigSecureServer failed");
  342. goto fail05;
  343. }
  344. }
  345. // initialize network
  346. if (!BNetwork_GlobalInit()) {
  347. BLog(BLOG_ERROR, "BNetwork_GlobalInit failed");
  348. goto fail1;
  349. }
  350. // process arguments
  351. if (!process_arguments()) {
  352. BLog(BLOG_ERROR, "Failed to process arguments");
  353. goto fail1;
  354. }
  355. // init communication predicate
  356. if (options.comm_predicate) {
  357. // init predicate
  358. if (!BPredicate_Init(&comm_predicate, options.comm_predicate)) {
  359. BLog(BLOG_ERROR, "BPredicate_Init failed");
  360. goto fail1;
  361. }
  362. // init functions
  363. int args[] = {PREDICATE_TYPE_STRING};
  364. BPredicateFunction_Init(&comm_predicate_func_p1name, &comm_predicate, "p1name", args, 1, comm_predicate_func_p1name_cb, NULL);
  365. BPredicateFunction_Init(&comm_predicate_func_p2name, &comm_predicate, "p2name", args, 1, comm_predicate_func_p2name_cb, NULL);
  366. BPredicateFunction_Init(&comm_predicate_func_p1addr, &comm_predicate, "p1addr", args, 1, comm_predicate_func_p1addr_cb, NULL);
  367. BPredicateFunction_Init(&comm_predicate_func_p2addr, &comm_predicate, "p2addr", args, 1, comm_predicate_func_p2addr_cb, NULL);
  368. }
  369. // init relay predicate
  370. if (options.relay_predicate) {
  371. // init predicate
  372. if (!BPredicate_Init(&relay_predicate, options.relay_predicate)) {
  373. BLog(BLOG_ERROR, "BPredicate_Init failed");
  374. goto fail2;
  375. }
  376. // init functions
  377. int args[] = {PREDICATE_TYPE_STRING};
  378. BPredicateFunction_Init(&relay_predicate_func_pname, &relay_predicate, "pname", args, 1, relay_predicate_func_pname_cb, NULL);
  379. BPredicateFunction_Init(&relay_predicate_func_rname, &relay_predicate, "rname", args, 1, relay_predicate_func_rname_cb, NULL);
  380. BPredicateFunction_Init(&relay_predicate_func_paddr, &relay_predicate, "paddr", args, 1, relay_predicate_func_paddr_cb, NULL);
  381. BPredicateFunction_Init(&relay_predicate_func_raddr, &relay_predicate, "raddr", args, 1, relay_predicate_func_raddr_cb, NULL);
  382. }
  383. // init time
  384. BTime_Init();
  385. // initialize reactor
  386. if (!BReactor_Init(&ss)) {
  387. BLog(BLOG_ERROR, "BReactor_Init failed");
  388. goto fail3;
  389. }
  390. // init thread work dispatcher
  391. if (!BThreadWorkDispatcher_Init(&twd, &ss, options.threads)) {
  392. BLog(BLOG_ERROR, "BThreadWorkDispatcher_Init failed");
  393. goto fail3a;
  394. }
  395. // setup signal handler
  396. if (!BSignal_Init(&ss, signal_handler, NULL)) {
  397. BLog(BLOG_ERROR, "BSignal_Init failed");
  398. goto fail4;
  399. }
  400. // initialize number of clients
  401. clients_num = 0;
  402. // first client ID will be zero
  403. clients_nextid = 0;
  404. // initialize clients linked list
  405. LinkedList1_Init(&clients);
  406. // initialize clients tree
  407. BAVL_Init(&clients_tree, OFFSET_DIFF(struct client_data, id, tree_node), (BAVL_comparator)peerid_comparator, NULL);
  408. // initialize listeners
  409. num_listeners = 0;
  410. while (num_listeners < num_listen_addrs) {
  411. if (!BListener_Init(&listeners[num_listeners], listen_addrs[num_listeners], &ss, &listeners[num_listeners], (BListener_handler)listener_handler)) {
  412. BLog(BLOG_ERROR, "BListener_Init failed");
  413. goto fail10;
  414. }
  415. num_listeners++;
  416. }
  417. // enter event loop
  418. BLog(BLOG_NOTICE, "entering event loop");
  419. BReactor_Exec(&ss);
  420. // free clients
  421. LinkedList1Node *node;
  422. while (node = LinkedList1_GetFirst(&clients)) {
  423. struct client_data *client = UPPER_OBJECT(node, struct client_data, list_node);
  424. // remove outgoing knows
  425. LinkedList1Node *node2;
  426. while (node2 = LinkedList1_GetFirst(&client->know_out_list)) {
  427. struct peer_know *k = UPPER_OBJECT(node2, struct peer_know, from_node);
  428. remove_know(k);
  429. }
  430. // remove incoming knows
  431. LinkedList1Node *node3;
  432. while (node3 = LinkedList1_GetFirst(&client->know_in_list)) {
  433. struct peer_know *k = UPPER_OBJECT(node3, struct peer_know, to_node);
  434. remove_know(k);
  435. }
  436. // remove outgoing flows
  437. LinkedList1Node *flow_node;
  438. while (flow_node = LinkedList1_GetFirst(&client->peer_out_flows_list)) {
  439. struct peer_flow *flow = UPPER_OBJECT(flow_node, struct peer_flow, src_list_node);
  440. ASSERT(flow->src_client == client)
  441. // allow freeing queue flows at dest
  442. PacketPassFairQueue_PrepareFree(&flow->dest_client->output_peers_fairqueue);
  443. // deallocate flow
  444. peer_flow_dealloc(flow);
  445. }
  446. // deallocate client
  447. client_dealloc(client);
  448. }
  449. fail10:
  450. while (num_listeners > 0) {
  451. num_listeners--;
  452. BListener_Free(&listeners[num_listeners]);
  453. }
  454. BSignal_Finish();
  455. fail4:
  456. BThreadWorkDispatcher_Free(&twd);
  457. fail3a:
  458. BReactor_Free(&ss);
  459. fail3:
  460. if (options.relay_predicate) {
  461. BPredicateFunction_Free(&relay_predicate_func_raddr);
  462. BPredicateFunction_Free(&relay_predicate_func_paddr);
  463. BPredicateFunction_Free(&relay_predicate_func_rname);
  464. BPredicateFunction_Free(&relay_predicate_func_pname);
  465. BPredicate_Free(&relay_predicate);
  466. }
  467. fail2:
  468. if (options.comm_predicate) {
  469. BPredicateFunction_Free(&comm_predicate_func_p2addr);
  470. BPredicateFunction_Free(&comm_predicate_func_p1addr);
  471. BPredicateFunction_Free(&comm_predicate_func_p2name);
  472. BPredicateFunction_Free(&comm_predicate_func_p1name);
  473. BPredicate_Free(&comm_predicate);
  474. }
  475. fail1:
  476. if (options.ssl) {
  477. fail05:
  478. ASSERT_FORCE(PR_Close(model_prfd) == PR_SUCCESS)
  479. fail04:
  480. CERT_DestroyCertificate(server_cert);
  481. SECKEY_DestroyPrivateKey(server_key);
  482. fail03:
  483. ASSERT_FORCE(SSL_ShutdownServerSessionIDCache() == SECSuccess)
  484. fail02:
  485. ASSERT_FORCE(NSS_Shutdown() == SECSuccess)
  486. fail01:
  487. ASSERT_FORCE(PR_Cleanup() == PR_SUCCESS)
  488. PL_ArenaFinish();
  489. }
  490. BLog(BLOG_NOTICE, "exiting");
  491. BLog_Free();
  492. fail0:
  493. DebugObjectGlobal_Finish();
  494. return 1;
  495. }
  496. void print_help (const char *name)
  497. {
  498. printf(
  499. "Usage:\n"
  500. " %s\n"
  501. " [--help]\n"
  502. " [--version]\n"
  503. " [--logger <"LOGGERS_STRING">]\n"
  504. #ifndef BADVPN_USE_WINAPI
  505. " (logger=syslog?\n"
  506. " [--syslog-facility <string>]\n"
  507. " [--syslog-ident <string>]\n"
  508. " )\n"
  509. #endif
  510. " [--loglevel <0-5/none/error/warning/notice/info/debug>]\n"
  511. " [--channel-loglevel <channel-name> <0-5/none/error/warning/notice/info/debug>] ...\n"
  512. " [--threads <integer>]\n"
  513. " [--use-threads-for-ssl-handshake]\n"
  514. " [--use-threads-for-ssl-data]\n"
  515. " [--listen-addr <addr>] ...\n"
  516. " [--ssl --nssdb <string> --server-cert-name <string>]\n"
  517. " [--comm-predicate <string>]\n"
  518. " [--relay-predicate <string>]\n"
  519. " [--client-socket-sndbuf <bytes / 0>]\n"
  520. " [--max-clients <number>]\n"
  521. "Address format is a.b.c.d:port (IPv4) or [addr]:port (IPv6).\n",
  522. name
  523. );
  524. }
  525. void print_version (void)
  526. {
  527. printf(GLOBAL_PRODUCT_NAME" "PROGRAM_NAME" "GLOBAL_VERSION"\n"GLOBAL_COPYRIGHT_NOTICE"\n");
  528. }
  529. int parse_arguments (int argc, char *argv[])
  530. {
  531. options.help = 0;
  532. options.version = 0;
  533. options.logger = LOGGER_STDOUT;
  534. #ifndef BADVPN_USE_WINAPI
  535. options.logger_syslog_facility = "daemon";
  536. options.logger_syslog_ident = argv[0];
  537. #endif
  538. options.loglevel = -1;
  539. for (int i = 0; i < BLOG_NUM_CHANNELS; i++) {
  540. options.loglevels[i] = -1;
  541. }
  542. options.threads = 0;
  543. options.use_threads_for_ssl_handshake = 0;
  544. options.use_threads_for_ssl_data = 0;
  545. options.ssl = 0;
  546. options.nssdb = NULL;
  547. options.server_cert_name = NULL;
  548. options.num_listen_addrs = 0;
  549. options.comm_predicate = NULL;
  550. options.relay_predicate = NULL;
  551. options.client_socket_sndbuf = CLIENT_DEFAULT_SOCKET_SNDBUF;
  552. options.max_clients = DEFAULT_MAX_CLIENTS;
  553. for (int i = 1; i < argc; i++) {
  554. char *arg = argv[i];
  555. if (!strcmp(arg, "--help")) {
  556. options.help = 1;
  557. }
  558. else if (!strcmp(arg, "--version")) {
  559. options.version = 1;
  560. }
  561. else if (!strcmp(arg, "--logger")) {
  562. if (i + 1 >= argc) {
  563. fprintf(stderr, "%s: requires an argument\n", arg);
  564. return 0;
  565. }
  566. char *arg2 = argv[i + 1];
  567. if (!strcmp(arg2, "stdout")) {
  568. options.logger = LOGGER_STDOUT;
  569. }
  570. #ifndef BADVPN_USE_WINAPI
  571. else if (!strcmp(arg2, "syslog")) {
  572. options.logger = LOGGER_SYSLOG;
  573. }
  574. #endif
  575. else {
  576. fprintf(stderr, "%s: wrong argument\n", arg);
  577. return 0;
  578. }
  579. i++;
  580. }
  581. #ifndef BADVPN_USE_WINAPI
  582. else if (!strcmp(arg, "--syslog-facility")) {
  583. if (i + 1 >= argc) {
  584. fprintf(stderr, "%s: requires an argument\n", arg);
  585. return 0;
  586. }
  587. options.logger_syslog_facility = argv[i + 1];
  588. i++;
  589. }
  590. else if (!strcmp(arg, "--syslog-ident")) {
  591. if (i + 1 >= argc) {
  592. fprintf(stderr, "%s: requires an argument\n", arg);
  593. return 0;
  594. }
  595. options.logger_syslog_ident = argv[i + 1];
  596. i++;
  597. }
  598. #endif
  599. else if (!strcmp(arg, "--loglevel")) {
  600. if (1 >= argc - i) {
  601. fprintf(stderr, "%s: requires an argument\n", arg);
  602. return 0;
  603. }
  604. if ((options.loglevel = parse_loglevel(argv[i + 1])) < 0) {
  605. fprintf(stderr, "%s: wrong argument\n", arg);
  606. return 0;
  607. }
  608. i++;
  609. }
  610. else if (!strcmp(arg, "--channel-loglevel")) {
  611. if (2 >= argc - i) {
  612. fprintf(stderr, "%s: requires two arguments\n", arg);
  613. return 0;
  614. }
  615. int channel = BLogGlobal_GetChannelByName(argv[i + 1]);
  616. if (channel < 0) {
  617. fprintf(stderr, "%s: wrong channel argument\n", arg);
  618. return 0;
  619. }
  620. int loglevel = parse_loglevel(argv[i + 2]);
  621. if (loglevel < 0) {
  622. fprintf(stderr, "%s: wrong loglevel argument\n", arg);
  623. return 0;
  624. }
  625. options.loglevels[channel] = loglevel;
  626. i += 2;
  627. }
  628. else if (!strcmp(arg, "--threads")) {
  629. if (1 >= argc - i) {
  630. fprintf(stderr, "%s: requires an argument\n", arg);
  631. return 0;
  632. }
  633. options.threads = atoi(argv[i + 1]);
  634. i++;
  635. }
  636. else if (!strcmp(arg, "--use-threads-for-ssl-handshake")) {
  637. options.use_threads_for_ssl_handshake = 1;
  638. }
  639. else if (!strcmp(arg, "--use-threads-for-ssl-data")) {
  640. options.use_threads_for_ssl_data = 1;
  641. }
  642. else if (!strcmp(arg, "--ssl")) {
  643. options.ssl = 1;
  644. }
  645. else if (!strcmp(arg, "--nssdb")) {
  646. if (1 >= argc - i) {
  647. fprintf(stderr, "%s: requires an argument\n", arg);
  648. return 0;
  649. }
  650. options.nssdb = argv[i + 1];
  651. i++;
  652. }
  653. else if (!strcmp(arg, "--server-cert-name")) {
  654. if (1 >= argc - i) {
  655. fprintf(stderr, "%s: requires an argument\n", arg);
  656. return 0;
  657. }
  658. options.server_cert_name = argv[i + 1];
  659. i++;
  660. }
  661. else if (!strcmp(arg, "--listen-addr")) {
  662. if (1 >= argc - i) {
  663. fprintf(stderr, "%s: requires an argument\n", arg);
  664. return 0;
  665. }
  666. if (options.num_listen_addrs == MAX_LISTEN_ADDRS) {
  667. fprintf(stderr, "%s: too many\n", arg);
  668. return 0;
  669. }
  670. options.listen_addrs[options.num_listen_addrs] = argv[i + 1];
  671. options.num_listen_addrs++;
  672. i++;
  673. }
  674. else if (!strcmp(arg, "--comm-predicate")) {
  675. if (1 >= argc - i) {
  676. fprintf(stderr, "%s: requires an argument\n", arg);
  677. return 0;
  678. }
  679. options.comm_predicate = argv[i + 1];
  680. i++;
  681. }
  682. else if (!strcmp(arg, "--relay-predicate")) {
  683. if (1 >= argc - i) {
  684. fprintf(stderr, "%s: requires an argument\n", arg);
  685. return 0;
  686. }
  687. options.relay_predicate = argv[i + 1];
  688. i++;
  689. }
  690. else if (!strcmp(arg, "--client-socket-sndbuf")) {
  691. if (1 >= argc - i) {
  692. fprintf(stderr, "%s: requires an argument\n", arg);
  693. return 0;
  694. }
  695. if ((options.client_socket_sndbuf = atoi(argv[i + 1])) < 0) {
  696. fprintf(stderr, "%s: wrong argument\n", arg);
  697. return 0;
  698. }
  699. i++;
  700. }
  701. else if (!strcmp(arg, "--max-clients")) {
  702. if (1 >= argc - i) {
  703. fprintf(stderr, "%s: requires an argument\n", arg);
  704. return 0;
  705. }
  706. if ((options.max_clients = atoi(argv[i + 1])) <= 0) {
  707. fprintf(stderr, "%s: wrong argument\n", arg);
  708. return 0;
  709. }
  710. i++;
  711. }
  712. else {
  713. fprintf(stderr, "%s: unknown option\n", arg);
  714. return 0;
  715. }
  716. }
  717. if (options.help || options.version) {
  718. return 1;
  719. }
  720. if (!!options.nssdb != options.ssl) {
  721. fprintf(stderr, "--ssl and --nssdb must be used together\n");
  722. return 0;
  723. }
  724. if (!!options.server_cert_name != options.ssl) {
  725. fprintf(stderr, "--ssl and --server-cert-name must be used together\n");
  726. return 0;
  727. }
  728. return 1;
  729. }
  730. int process_arguments (void)
  731. {
  732. // resolve listen addresses
  733. num_listen_addrs = 0;
  734. while (num_listen_addrs < options.num_listen_addrs) {
  735. if (!BAddr_Parse(&listen_addrs[num_listen_addrs], options.listen_addrs[num_listen_addrs], NULL, 0)) {
  736. BLog(BLOG_ERROR, "listen addr: BAddr_Parse failed");
  737. return 0;
  738. }
  739. num_listen_addrs++;
  740. }
  741. return 1;
  742. }
  743. int ssl_flags (void)
  744. {
  745. int flags = 0;
  746. if (options.use_threads_for_ssl_handshake) {
  747. flags |= BSSLCONNECTION_FLAG_THREADWORK_HANDSHAKE;
  748. }
  749. if (options.use_threads_for_ssl_data) {
  750. flags |= BSSLCONNECTION_FLAG_THREADWORK_IO;
  751. }
  752. return flags;
  753. }
  754. void signal_handler (void *unused)
  755. {
  756. BLog(BLOG_NOTICE, "termination requested");
  757. // exit event loop
  758. BReactor_Quit(&ss, 0);
  759. }
  760. void listener_handler (BListener *listener)
  761. {
  762. if (clients_num == options.max_clients) {
  763. BLog(BLOG_WARNING, "too many clients for new client");
  764. goto fail0;
  765. }
  766. // allocate the client structure
  767. struct client_data *client = (struct client_data *)malloc(sizeof(*client));
  768. if (!client) {
  769. BLog(BLOG_ERROR, "failed to allocate client");
  770. goto fail0;
  771. }
  772. // accept connection
  773. if (!BConnection_Init(&client->con, BConnection_source_listener(listener, &client->addr), &ss, client, (BConnection_handler)client_connection_handler)) {
  774. BLog(BLOG_ERROR, "BConnection_Init failed");
  775. goto fail1;
  776. }
  777. // limit socket send buffer, else our scheduling is pointless
  778. if (options.client_socket_sndbuf > 0) {
  779. if (!BConnection_SetSendBuffer(&client->con, options.client_socket_sndbuf)) {
  780. BLog(BLOG_WARNING, "BConnection_SetSendBuffer failed");
  781. }
  782. }
  783. // assign ID
  784. client->id = new_client_id();
  785. // set no common name
  786. client->common_name = NULL;
  787. // now client_log() works
  788. // init connection interfaces
  789. BConnection_SendAsync_Init(&client->con);
  790. BConnection_RecvAsync_Init(&client->con);
  791. if (options.ssl) {
  792. // create bottom NSPR file descriptor
  793. if (!BSSLConnection_MakeBackend(&client->bottom_prfd, BConnection_SendAsync_GetIf(&client->con), BConnection_RecvAsync_GetIf(&client->con), &twd, ssl_flags())) {
  794. client_log(client, BLOG_ERROR, "BSSLConnection_MakeBackend failed");
  795. goto fail2;
  796. }
  797. // create SSL file descriptor from the bottom NSPR file descriptor
  798. if (!(client->ssl_prfd = SSL_ImportFD(model_prfd, &client->bottom_prfd))) {
  799. client_log(client, BLOG_ERROR, "SSL_ImportFD failed");
  800. ASSERT_FORCE(PR_Close(&client->bottom_prfd) == PR_SUCCESS)
  801. goto fail2;
  802. }
  803. // set server mode
  804. if (SSL_ResetHandshake(client->ssl_prfd, PR_TRUE) != SECSuccess) {
  805. client_log(client, BLOG_ERROR, "SSL_ResetHandshake failed");
  806. goto fail3;
  807. }
  808. // set require client certificate
  809. if (SSL_OptionSet(client->ssl_prfd, SSL_REQUEST_CERTIFICATE, PR_TRUE) != SECSuccess) {
  810. client_log(client, BLOG_ERROR, "SSL_OptionSet(SSL_REQUEST_CERTIFICATE) failed");
  811. goto fail3;
  812. }
  813. if (SSL_OptionSet(client->ssl_prfd, SSL_REQUIRE_CERTIFICATE, PR_TRUE) != SECSuccess) {
  814. client_log(client, BLOG_ERROR, "SSL_OptionSet(SSL_REQUIRE_CERTIFICATE) failed");
  815. goto fail3;
  816. }
  817. // init SSL connection
  818. BSSLConnection_Init(&client->sslcon, client->ssl_prfd, 1, BReactor_PendingGroup(&ss), client, (BSSLConnection_handler)client_sslcon_handler);
  819. } else {
  820. // initialize I/O
  821. if (!client_init_io(client)) {
  822. goto fail2;
  823. }
  824. }
  825. // start disconnect timer
  826. BTimer_Init(&client->disconnect_timer, CLIENT_NO_DATA_TIME_LIMIT, (BTimer_handler)client_disconnect_timer_handler, client);
  827. BReactor_SetTimer(&ss, &client->disconnect_timer);
  828. // link in
  829. clients_num++;
  830. LinkedList1_Append(&clients, &client->list_node);
  831. ASSERT_EXECUTE(BAVL_Insert(&clients_tree, &client->tree_node, NULL))
  832. // init knowledge lists
  833. LinkedList1_Init(&client->know_out_list);
  834. LinkedList1_Init(&client->know_in_list);
  835. // initialize peer flows from us list and tree (flows for sending messages to other clients)
  836. LinkedList1_Init(&client->peer_out_flows_list);
  837. BAVL_Init(&client->peer_out_flows_tree, OFFSET_DIFF(struct peer_flow, dest_client_id, src_tree_node), (BAVL_comparator)peerid_comparator, NULL);
  838. // init dying
  839. client->dying = 0;
  840. BPending_Init(&client->dying_job, BReactor_PendingGroup(&ss), (BPending_handler)client_dying_job, client);
  841. // set state
  842. client->initstatus = (options.ssl ? INITSTATUS_HANDSHAKE : INITSTATUS_WAITHELLO);
  843. client_log(client, BLOG_INFO, "initialized");
  844. return;
  845. if (options.ssl) {
  846. fail3:
  847. ASSERT_FORCE(PR_Close(client->ssl_prfd) == PR_SUCCESS)
  848. }
  849. fail2:
  850. BConnection_RecvAsync_Free(&client->con);
  851. BConnection_SendAsync_Free(&client->con);
  852. BConnection_Free(&client->con);
  853. fail1:
  854. free(client);
  855. fail0:
  856. return;
  857. }
  858. void client_dealloc (struct client_data *client)
  859. {
  860. ASSERT(LinkedList1_IsEmpty(&client->know_out_list))
  861. ASSERT(LinkedList1_IsEmpty(&client->know_in_list))
  862. ASSERT(LinkedList1_IsEmpty(&client->peer_out_flows_list))
  863. // free I/O
  864. if (client->initstatus >= INITSTATUS_WAITHELLO && !client->dying) {
  865. client_dealloc_io(client);
  866. }
  867. // free dying
  868. BPending_Free(&client->dying_job);
  869. // link out
  870. BAVL_Remove(&clients_tree, &client->tree_node);
  871. LinkedList1_Remove(&clients, &client->list_node);
  872. clients_num--;
  873. // stop disconnect timer
  874. BReactor_RemoveTimer(&ss, &client->disconnect_timer);
  875. // free SSL
  876. if (options.ssl) {
  877. BSSLConnection_Free(&client->sslcon);
  878. ASSERT_FORCE(PR_Close(client->ssl_prfd) == PR_SUCCESS)
  879. }
  880. // free common name
  881. if (client->common_name) {
  882. PORT_Free(client->common_name);
  883. }
  884. // free connection interfaces
  885. BConnection_RecvAsync_Free(&client->con);
  886. BConnection_SendAsync_Free(&client->con);
  887. // free connection
  888. BConnection_Free(&client->con);
  889. // free memory
  890. free(client);
  891. }
  892. int client_compute_buffer_size (struct client_data *client)
  893. {
  894. bsize_t s = bsize_add(bsize_fromsize(1), bsize_mul(bsize_fromsize(2), bsize_fromsize(options.max_clients - 1)));
  895. if (s.is_overflow || s.value > INT_MAX) {
  896. return INT_MAX;
  897. } else {
  898. return s.value;
  899. }
  900. }
  901. int client_init_io (struct client_data *client)
  902. {
  903. StreamPassInterface *send_if = (options.ssl ? BSSLConnection_GetSendIf(&client->sslcon) : BConnection_SendAsync_GetIf(&client->con));
  904. StreamRecvInterface *recv_if = (options.ssl ? BSSLConnection_GetRecvIf(&client->sslcon) : BConnection_RecvAsync_GetIf(&client->con));
  905. // init input
  906. // init interface
  907. PacketPassInterface_Init(&client->input_interface, SC_MAX_ENC, (PacketPassInterface_handler_send)client_input_handler_send, client, BReactor_PendingGroup(&ss));
  908. // init decoder
  909. if (!PacketProtoDecoder_Init(&client->input_decoder, recv_if, &client->input_interface, BReactor_PendingGroup(&ss), client,
  910. (PacketProtoDecoder_handler_error)client_decoder_handler_error
  911. )) {
  912. client_log(client, BLOG_ERROR, "PacketProtoDecoder_Init failed");
  913. goto fail1;
  914. }
  915. // init output common
  916. // init sender
  917. PacketStreamSender_Init(&client->output_sender, send_if, PACKETPROTO_ENCLEN(SC_MAX_ENC), BReactor_PendingGroup(&ss));
  918. // init queue
  919. PacketPassPriorityQueue_Init(&client->output_priorityqueue, PacketStreamSender_GetInput(&client->output_sender), BReactor_PendingGroup(&ss), 0);
  920. // init output control flow
  921. // init queue flow
  922. PacketPassPriorityQueueFlow_Init(&client->output_control_qflow, &client->output_priorityqueue, -1);
  923. // init PacketProtoFlow
  924. if (!PacketProtoFlow_Init(
  925. &client->output_control_oflow, SC_MAX_ENC, client_compute_buffer_size(client),
  926. PacketPassPriorityQueueFlow_GetInput(&client->output_control_qflow), BReactor_PendingGroup(&ss)
  927. )) {
  928. client_log(client, BLOG_ERROR, "PacketProtoFlow_Init failed");
  929. goto fail2;
  930. }
  931. client->output_control_input = PacketProtoFlow_GetInput(&client->output_control_oflow);
  932. client->output_control_packet_len = -1;
  933. // init output peers flow
  934. // init queue flow
  935. // use lower priority than control flow (higher number)
  936. PacketPassPriorityQueueFlow_Init(&client->output_peers_qflow, &client->output_priorityqueue, 0);
  937. // init fair queue (for different peers)
  938. if (!PacketPassFairQueue_Init(&client->output_peers_fairqueue, PacketPassPriorityQueueFlow_GetInput(&client->output_peers_qflow), BReactor_PendingGroup(&ss), 0, 1)) {
  939. client_log(client, BLOG_ERROR, "PacketPassFairQueue_Init failed");
  940. goto fail3;
  941. }
  942. // init list of flows
  943. LinkedList1_Init(&client->output_peers_flows);
  944. return 1;
  945. fail3:
  946. PacketPassPriorityQueueFlow_Free(&client->output_peers_qflow);
  947. PacketProtoFlow_Free(&client->output_control_oflow);
  948. fail2:
  949. PacketPassPriorityQueueFlow_Free(&client->output_control_qflow);
  950. // free output common
  951. PacketPassPriorityQueue_Free(&client->output_priorityqueue);
  952. PacketStreamSender_Free(&client->output_sender);
  953. // free input
  954. PacketProtoDecoder_Free(&client->input_decoder);
  955. fail1:
  956. PacketPassInterface_Free(&client->input_interface);
  957. return 0;
  958. }
  959. void client_dealloc_io (struct client_data *client)
  960. {
  961. // stop using any buffers before they get freed
  962. if (options.ssl) {
  963. BSSLConnection_ReleaseBuffers(&client->sslcon);
  964. }
  965. // allow freeing fair queue flows
  966. PacketPassFairQueue_PrepareFree(&client->output_peers_fairqueue);
  967. // remove flows to us
  968. LinkedList1Node *node;
  969. while (node = LinkedList1_GetFirst(&client->output_peers_flows)) {
  970. struct peer_flow *flow = UPPER_OBJECT(node, struct peer_flow, dest_list_node);
  971. ASSERT(flow->dest_client == client)
  972. peer_flow_dealloc(flow);
  973. }
  974. // allow freeing priority queue flows
  975. PacketPassPriorityQueue_PrepareFree(&client->output_priorityqueue);
  976. // free output peers flow
  977. PacketPassFairQueue_Free(&client->output_peers_fairqueue);
  978. PacketPassPriorityQueueFlow_Free(&client->output_peers_qflow);
  979. // free output control flow
  980. PacketProtoFlow_Free(&client->output_control_oflow);
  981. PacketPassPriorityQueueFlow_Free(&client->output_control_qflow);
  982. // free output common
  983. PacketPassPriorityQueue_Free(&client->output_priorityqueue);
  984. PacketStreamSender_Free(&client->output_sender);
  985. // free input
  986. PacketProtoDecoder_Free(&client->input_decoder);
  987. PacketPassInterface_Free(&client->input_interface);
  988. }
  989. void client_remove (struct client_data *client)
  990. {
  991. ASSERT(!client->dying)
  992. client_log(client, BLOG_INFO, "removing");
  993. // set dying to prevent sending this client anything
  994. client->dying = 1;
  995. // free I/O now, removing incoming flows
  996. if (client->initstatus >= INITSTATUS_WAITHELLO) {
  997. client_dealloc_io(client);
  998. }
  999. // remove outgoing knows
  1000. LinkedList1Node *node;
  1001. while (node = LinkedList1_GetFirst(&client->know_out_list)) {
  1002. struct peer_know *k = UPPER_OBJECT(node, struct peer_know, from_node);
  1003. remove_know(k);
  1004. }
  1005. // remove outgoing flows
  1006. while (node = LinkedList1_GetFirst(&client->peer_out_flows_list)) {
  1007. struct peer_flow *flow = UPPER_OBJECT(node, struct peer_flow, src_list_node);
  1008. ASSERT(flow->src_client == client)
  1009. ASSERT(flow->dest_client->initstatus == INITSTATUS_COMPLETE)
  1010. ASSERT(!flow->dest_client->dying)
  1011. if (flow->have_io && PacketPassFairQueueFlow_IsBusy(&flow->qflow)) {
  1012. client_log(client, BLOG_DEBUG, "removing flow to %d later", (int)flow->dest_client->id);
  1013. peer_flow_disconnect(flow);
  1014. } else {
  1015. client_log(client, BLOG_DEBUG, "removing flow to %d now", (int)flow->dest_client->id);
  1016. peer_flow_dealloc(flow);
  1017. }
  1018. }
  1019. // schedule job to finish removal after clients are informed
  1020. BPending_Set(&client->dying_job);
  1021. // inform other clients that 'client' is no more
  1022. node = LinkedList1_GetFirst(&client->know_in_list);
  1023. while (node) {
  1024. LinkedList1Node *next = LinkedList1Node_Next(node);
  1025. struct peer_know *k = UPPER_OBJECT(node, struct peer_know, to_node);
  1026. uninform_know(k);
  1027. node = next;
  1028. }
  1029. }
  1030. void client_dying_job (struct client_data *client)
  1031. {
  1032. ASSERT(client->dying)
  1033. ASSERT(LinkedList1_IsEmpty(&client->know_in_list))
  1034. client_dealloc(client);
  1035. return;
  1036. }
  1037. void client_logfunc (struct client_data *client)
  1038. {
  1039. char addr[BADDR_MAX_PRINT_LEN];
  1040. BAddr_Print(&client->addr, addr);
  1041. BLog_Append("client %d (%s)", (int)client->id, addr);
  1042. if (client->common_name) {
  1043. BLog_Append(" (%s)", client->common_name);
  1044. }
  1045. BLog_Append(": ");
  1046. }
  1047. void client_log (struct client_data *client, int level, const char *fmt, ...)
  1048. {
  1049. va_list vl;
  1050. va_start(vl, fmt);
  1051. BLog_LogViaFuncVarArg((BLog_logfunc)client_logfunc, client, BLOG_CURRENT_CHANNEL, level, fmt, vl);
  1052. va_end(vl);
  1053. }
  1054. void client_disconnect_timer_handler (struct client_data *client)
  1055. {
  1056. ASSERT(!client->dying)
  1057. client_log(client, BLOG_INFO, "timed out");
  1058. client_remove(client);
  1059. return;
  1060. }
  1061. void client_connection_handler (struct client_data *client, int event)
  1062. {
  1063. ASSERT(!client->dying)
  1064. if (event == BCONNECTION_EVENT_RECVCLOSED) {
  1065. client_log(client, BLOG_INFO, "connection closed");
  1066. } else {
  1067. client_log(client, BLOG_INFO, "connection error");
  1068. }
  1069. client_remove(client);
  1070. return;
  1071. }
  1072. void client_sslcon_handler (struct client_data *client, int event)
  1073. {
  1074. ASSERT(options.ssl)
  1075. ASSERT(!client->dying)
  1076. ASSERT(event == BSSLCONNECTION_EVENT_UP || event == BSSLCONNECTION_EVENT_ERROR)
  1077. ASSERT(!(event == BSSLCONNECTION_EVENT_UP) || client->initstatus == INITSTATUS_HANDSHAKE)
  1078. if (event == BSSLCONNECTION_EVENT_ERROR) {
  1079. client_log(client, BLOG_ERROR, "SSL error");
  1080. client_remove(client);
  1081. return;
  1082. }
  1083. // get client certificate
  1084. CERTCertificate *cert = SSL_PeerCertificate(client->ssl_prfd);
  1085. if (!cert) {
  1086. client_log(client, BLOG_ERROR, "SSL_PeerCertificate failed");
  1087. goto fail0;
  1088. }
  1089. // remember common name
  1090. if (!(client->common_name = CERT_GetCommonName(&cert->subject))) {
  1091. client_log(client, BLOG_NOTICE, "CERT_GetCommonName failed");
  1092. goto fail1;
  1093. }
  1094. // store certificate
  1095. SECItem der = cert->derCert;
  1096. if (der.len > sizeof(client->cert)) {
  1097. client_log(client, BLOG_NOTICE, "client certificate too big");
  1098. goto fail1;
  1099. }
  1100. memcpy(client->cert, der.data, der.len);
  1101. client->cert_len = der.len;
  1102. PRArenaPool *arena = PORT_NewArena(DER_DEFAULT_CHUNKSIZE);
  1103. if (!arena) {
  1104. client_log(client, BLOG_ERROR, "PORT_NewArena failed");
  1105. goto fail1;
  1106. }
  1107. // encode certificate
  1108. memset(&der, 0, sizeof(der));
  1109. if (!SEC_ASN1EncodeItem(arena, &der, cert, SEC_ASN1_GET(CERT_CertificateTemplate))) {
  1110. client_log(client, BLOG_ERROR, "SEC_ASN1EncodeItem failed");
  1111. goto fail2;
  1112. }
  1113. // store re-encoded certificate (for compatibility with old clients)
  1114. if (der.len > sizeof(client->cert_old)) {
  1115. client_log(client, BLOG_NOTICE, "client certificate too big");
  1116. goto fail2;
  1117. }
  1118. memcpy(client->cert_old, der.data, der.len);
  1119. client->cert_old_len = der.len;
  1120. // init I/O chains
  1121. if (!client_init_io(client)) {
  1122. goto fail2;
  1123. }
  1124. PORT_FreeArena(arena, PR_FALSE);
  1125. CERT_DestroyCertificate(cert);
  1126. // set client state
  1127. client->initstatus = INITSTATUS_WAITHELLO;
  1128. client_log(client, BLOG_INFO, "handshake complete");
  1129. return;
  1130. // handle errors
  1131. fail2:
  1132. PORT_FreeArena(arena, PR_FALSE);
  1133. fail1:
  1134. CERT_DestroyCertificate(cert);
  1135. fail0:
  1136. client_remove(client);
  1137. }
  1138. void client_decoder_handler_error (struct client_data *client)
  1139. {
  1140. ASSERT(INITSTATUS_HASLINK(client->initstatus))
  1141. ASSERT(!client->dying)
  1142. client_log(client, BLOG_ERROR, "decoder error");
  1143. client_remove(client);
  1144. return;
  1145. }
  1146. int client_start_control_packet (struct client_data *client, void **data, int len)
  1147. {
  1148. ASSERT(len >= 0)
  1149. ASSERT(len <= SC_MAX_PAYLOAD)
  1150. ASSERT(!(len > 0) || data)
  1151. ASSERT(INITSTATUS_HASLINK(client->initstatus))
  1152. ASSERT(!client->dying)
  1153. ASSERT(client->output_control_packet_len == -1)
  1154. #ifdef SIMULATE_OUT_OF_CONTROL_BUFFER
  1155. uint8_t x;
  1156. BRandom_randomize(&x, sizeof(x));
  1157. if (x < SIMULATE_OUT_OF_CONTROL_BUFFER) {
  1158. client_log(client, BLOG_INFO, "out of control buffer, removing");
  1159. client_remove(client);
  1160. return -1;
  1161. }
  1162. #endif
  1163. // obtain location for writing the packet
  1164. if (!BufferWriter_StartPacket(client->output_control_input, &client->output_control_packet)) {
  1165. // out of buffer, kill client
  1166. client_log(client, BLOG_INFO, "out of control buffer, removing");
  1167. client_remove(client);
  1168. return -1;
  1169. }
  1170. client->output_control_packet_len = len;
  1171. if (data) {
  1172. *data = client->output_control_packet + sizeof(struct sc_header);
  1173. }
  1174. return 0;
  1175. }
  1176. void client_end_control_packet (struct client_data *client, uint8_t type)
  1177. {
  1178. ASSERT(INITSTATUS_HASLINK(client->initstatus))
  1179. ASSERT(!client->dying)
  1180. ASSERT(client->output_control_packet_len >= 0)
  1181. ASSERT(client->output_control_packet_len <= SC_MAX_PAYLOAD)
  1182. // write header
  1183. struct sc_header header;
  1184. header.type = htol8(type);
  1185. memcpy(client->output_control_packet, &header, sizeof(header));
  1186. // finish writing packet
  1187. BufferWriter_EndPacket(client->output_control_input, sizeof(struct sc_header) + client->output_control_packet_len);
  1188. client->output_control_packet_len = -1;
  1189. }
  1190. int client_send_newclient (struct client_data *client, struct client_data *nc, int relay_server, int relay_client)
  1191. {
  1192. ASSERT(client->initstatus == INITSTATUS_COMPLETE)
  1193. ASSERT(!client->dying)
  1194. ASSERT(nc->initstatus == INITSTATUS_COMPLETE)
  1195. ASSERT(!nc->dying)
  1196. int flags = 0;
  1197. if (relay_server) {
  1198. flags |= SCID_NEWCLIENT_FLAG_RELAY_SERVER;
  1199. }
  1200. if (relay_client) {
  1201. flags |= SCID_NEWCLIENT_FLAG_RELAY_CLIENT;
  1202. }
  1203. if (options.ssl && client->version > SC_OLDVERSION_NOSSL && nc->version > SC_OLDVERSION_NOSSL) {
  1204. flags |= SCID_NEWCLIENT_FLAG_SSL;
  1205. }
  1206. uint8_t *cert_data = NULL;
  1207. int cert_len = 0;
  1208. if (options.ssl) {
  1209. cert_data = (client->version == SC_OLDVERSION_BROKENCERT ? nc->cert_old : nc->cert);
  1210. cert_len = (client->version == SC_OLDVERSION_BROKENCERT ? nc->cert_old_len : nc->cert_len);
  1211. }
  1212. struct sc_server_newclient omsg;
  1213. void *pack;
  1214. if (client_start_control_packet(client, &pack, sizeof(omsg) + cert_len) < 0) {
  1215. return -1;
  1216. }
  1217. omsg.id = htol16(nc->id);
  1218. omsg.flags = htol16(flags);
  1219. memcpy(pack, &omsg, sizeof(omsg));
  1220. if (cert_len > 0) {
  1221. memcpy((char *)pack + sizeof(omsg), cert_data, cert_len);
  1222. }
  1223. client_end_control_packet(client, SCID_NEWCLIENT);
  1224. return 0;
  1225. }
  1226. int client_send_endclient (struct client_data *client, peerid_t end_id)
  1227. {
  1228. ASSERT(client->initstatus == INITSTATUS_COMPLETE)
  1229. ASSERT(!client->dying)
  1230. struct sc_server_endclient omsg;
  1231. void *pack;
  1232. if (client_start_control_packet(client, &pack, sizeof(omsg)) < 0) {
  1233. return -1;
  1234. }
  1235. omsg.id = htol16(end_id);
  1236. memcpy(pack, &omsg, sizeof(omsg));
  1237. client_end_control_packet(client, SCID_ENDCLIENT);
  1238. return 0;
  1239. }
  1240. void client_input_handler_send (struct client_data *client, uint8_t *data, int data_len)
  1241. {
  1242. ASSERT(data_len >= 0)
  1243. ASSERT(data_len <= SC_MAX_ENC)
  1244. ASSERT(INITSTATUS_HASLINK(client->initstatus))
  1245. ASSERT(!client->dying)
  1246. // accept packet
  1247. PacketPassInterface_Done(&client->input_interface);
  1248. // restart disconnect timer
  1249. BReactor_SetTimer(&ss, &client->disconnect_timer);
  1250. // parse header
  1251. if (data_len < sizeof(struct sc_header)) {
  1252. client_log(client, BLOG_NOTICE, "packet too short");
  1253. client_remove(client);
  1254. return;
  1255. }
  1256. struct sc_header header;
  1257. memcpy(&header, data, sizeof(header));
  1258. data += sizeof(header);
  1259. data_len -= sizeof(header);
  1260. uint8_t type = ltoh8(header.type);
  1261. ASSERT(data_len >= 0)
  1262. ASSERT(data_len <= SC_MAX_PAYLOAD)
  1263. // perform action based on packet type
  1264. switch (type) {
  1265. case SCID_KEEPALIVE:
  1266. client_log(client, BLOG_DEBUG, "received keep-alive");
  1267. return;
  1268. case SCID_CLIENTHELLO:
  1269. process_packet_hello(client, data, data_len);
  1270. return;
  1271. case SCID_OUTMSG:
  1272. process_packet_outmsg(client, data, data_len);
  1273. return;
  1274. case SCID_RESETPEER:
  1275. process_packet_resetpeer(client, data, data_len);
  1276. return;
  1277. case SCID_ACCEPTPEER:
  1278. process_packet_acceptpeer(client, data, data_len);
  1279. return;
  1280. default:
  1281. client_log(client, BLOG_NOTICE, "unknown packet type %d, removing", (int)type);
  1282. client_remove(client);
  1283. return;
  1284. }
  1285. }
  1286. void process_packet_hello (struct client_data *client, uint8_t *data, int data_len)
  1287. {
  1288. if (client->initstatus != INITSTATUS_WAITHELLO) {
  1289. client_log(client, BLOG_NOTICE, "hello: not expected");
  1290. client_remove(client);
  1291. return;
  1292. }
  1293. if (data_len != sizeof(struct sc_client_hello)) {
  1294. client_log(client, BLOG_NOTICE, "hello: invalid length");
  1295. client_remove(client);
  1296. return;
  1297. }
  1298. struct sc_client_hello msg;
  1299. memcpy(&msg, data, sizeof(msg));
  1300. client->version = ltoh16(msg.version);
  1301. switch (client->version) {
  1302. case SC_VERSION:
  1303. case SC_OLDVERSION_NOSSL:
  1304. case SC_OLDVERSION_BROKENCERT:
  1305. break;
  1306. default:
  1307. client_log(client, BLOG_ERROR, "hello: unknown version (%d)", client->version);
  1308. client_remove(client);
  1309. return;
  1310. }
  1311. client_log(client, BLOG_INFO, "received hello");
  1312. // set client state to complete
  1313. client->initstatus = INITSTATUS_COMPLETE;
  1314. // publish client
  1315. for (LinkedList1Node *list_node = LinkedList1_GetFirst(&clients); list_node; list_node = LinkedList1Node_Next(list_node)) {
  1316. struct client_data *client2 = UPPER_OBJECT(list_node, struct client_data, list_node);
  1317. if (client2 == client || client2->initstatus != INITSTATUS_COMPLETE || client2->dying || !clients_allowed(client, client2)) {
  1318. continue;
  1319. }
  1320. // create flow from client to client2
  1321. struct peer_flow *flow_to = peer_flow_create(client, client2);
  1322. if (!flow_to) {
  1323. client_log(client, BLOG_ERROR, "failed to allocate flow to %d", (int)client2->id);
  1324. goto fail;
  1325. }
  1326. // create flow from client2 to client
  1327. struct peer_flow *flow_from = peer_flow_create(client2, client);
  1328. if (!flow_from) {
  1329. client_log(client, BLOG_ERROR, "failed to allocate flow from %d", (int)client2->id);
  1330. goto fail;
  1331. }
  1332. // set opposite flow pointers
  1333. flow_to->opposite = flow_from;
  1334. flow_from->opposite = flow_to;
  1335. // launch pair
  1336. if (!launch_pair(flow_to)) {
  1337. return;
  1338. }
  1339. }
  1340. // send hello
  1341. struct sc_server_hello omsg;
  1342. void *pack;
  1343. if (client_start_control_packet(client, &pack, sizeof(omsg)) < 0) {
  1344. return;
  1345. }
  1346. omsg.flags = htol16(0);
  1347. omsg.id = htol16(client->id);
  1348. omsg.clientAddr = (client->addr.type == BADDR_TYPE_IPV4 ? client->addr.ipv4.ip : hton32(0));
  1349. memcpy(pack, &omsg, sizeof(omsg));
  1350. client_end_control_packet(client, SCID_SERVERHELLO);
  1351. return;
  1352. fail:
  1353. client_remove(client);
  1354. }
  1355. void process_packet_outmsg (struct client_data *client, uint8_t *data, int data_len)
  1356. {
  1357. if (client->initstatus != INITSTATUS_COMPLETE) {
  1358. client_log(client, BLOG_NOTICE, "outmsg: not expected");
  1359. client_remove(client);
  1360. return;
  1361. }
  1362. if (data_len < sizeof(struct sc_client_outmsg)) {
  1363. client_log(client, BLOG_NOTICE, "outmsg: wrong size");
  1364. client_remove(client);
  1365. return;
  1366. }
  1367. struct sc_client_outmsg msg;
  1368. memcpy(&msg, data, sizeof(msg));
  1369. peerid_t id = ltoh16(msg.clientid);
  1370. int payload_size = data_len - sizeof(struct sc_client_outmsg);
  1371. if (payload_size > SC_MAX_MSGLEN) {
  1372. client_log(client, BLOG_NOTICE, "outmsg: too large payload");
  1373. client_remove(client);
  1374. return;
  1375. }
  1376. uint8_t *payload = data + sizeof(struct sc_client_outmsg);
  1377. // lookup flow to destination client
  1378. struct peer_flow *flow = find_flow(client, id);
  1379. if (!flow) {
  1380. client_log(client, BLOG_INFO, "no flow for message to %d", (int)id);
  1381. return;
  1382. }
  1383. // if pair is resetting, ignore message
  1384. if (flow->resetting || flow->opposite->resetting) {
  1385. client_log(client, BLOG_INFO, "pair is resetting; not forwarding message to %d", (int)id);
  1386. return;
  1387. }
  1388. // if sending client hasn't accepted yet, ignore message
  1389. if (!flow->accepted) {
  1390. client_log(client, BLOG_INFO, "client hasn't accepted; not forwarding message to %d", (int)id);
  1391. return;
  1392. }
  1393. #ifdef SIMULATE_OUT_OF_FLOW_BUFFER
  1394. uint8_t x;
  1395. BRandom_randomize(&x, sizeof(x));
  1396. if (x < SIMULATE_OUT_OF_FLOW_BUFFER) {
  1397. client_log(client, BLOG_WARNING, "simulating error; resetting to %d", (int)flow->dest_client->id);
  1398. peer_flow_start_reset(flow);
  1399. return;
  1400. }
  1401. #endif
  1402. // send packet
  1403. struct sc_server_inmsg omsg;
  1404. void *pack;
  1405. if (!peer_flow_start_packet(flow, &pack, sizeof(omsg) + payload_size)) {
  1406. // out of buffer, reset these two clients
  1407. client_log(client, BLOG_WARNING, "out of buffer; resetting to %d", (int)flow->dest_client->id);
  1408. peer_flow_start_reset(flow);
  1409. return;
  1410. }
  1411. omsg.clientid = htol16(client->id);
  1412. memcpy(pack, &omsg, sizeof(omsg));
  1413. memcpy((char *)pack + sizeof(omsg), payload, payload_size);
  1414. peer_flow_end_packet(flow, SCID_INMSG);
  1415. }
  1416. void process_packet_resetpeer (struct client_data *client, uint8_t *data, int data_len)
  1417. {
  1418. if (client->initstatus != INITSTATUS_COMPLETE) {
  1419. client_log(client, BLOG_NOTICE, "resetpeer: not expected");
  1420. client_remove(client);
  1421. return;
  1422. }
  1423. if (data_len != sizeof(struct sc_client_resetpeer)) {
  1424. client_log(client, BLOG_NOTICE, "resetpeer: wrong size");
  1425. client_remove(client);
  1426. return;
  1427. }
  1428. struct sc_client_resetpeer msg;
  1429. memcpy(&msg, data, sizeof(msg));
  1430. peerid_t id = ltoh16(msg.clientid);
  1431. // lookup flow to destination client
  1432. struct peer_flow *flow = find_flow(client, id);
  1433. if (!flow) {
  1434. client_log(client, BLOG_INFO, "no flow for reset to %d", (int)id);
  1435. return;
  1436. }
  1437. // if pair is resetting, ignore message
  1438. if (flow->resetting || flow->opposite->resetting) {
  1439. client_log(client, BLOG_INFO, "pair is resetting; not resetting to %d", (int)id);
  1440. return;
  1441. }
  1442. // if sending client hasn't accepted yet, ignore message
  1443. if (!flow->accepted) {
  1444. client_log(client, BLOG_INFO, "client hasn't accepted; not resetting to %d", (int)id);
  1445. return;
  1446. }
  1447. client_log(client, BLOG_WARNING, "resetting to %d", (int)flow->dest_client->id);
  1448. // reset clients
  1449. peer_flow_start_reset(flow);
  1450. }
  1451. void process_packet_acceptpeer (struct client_data *client, uint8_t *data, int data_len)
  1452. {
  1453. if (client->initstatus != INITSTATUS_COMPLETE) {
  1454. client_log(client, BLOG_NOTICE, "acceptpeer: not expected");
  1455. client_remove(client);
  1456. return;
  1457. }
  1458. if (data_len != sizeof(struct sc_client_acceptpeer)) {
  1459. client_log(client, BLOG_NOTICE, "acceptpeer: wrong size");
  1460. client_remove(client);
  1461. return;
  1462. }
  1463. struct sc_client_acceptpeer msg;
  1464. memcpy(&msg, data, sizeof(msg));
  1465. peerid_t id = ltoh16(msg.clientid);
  1466. // lookup flow to destination client
  1467. struct peer_flow *flow = find_flow(client, id);
  1468. if (!flow) {
  1469. // the specified client has probably gone away but the sending client didn't know
  1470. // that yet; this is expected
  1471. client_log(client, BLOG_INFO, "acceptpeer: no flow to %d", (int)id);
  1472. return;
  1473. }
  1474. // client can only accept once
  1475. if (flow->accepted) {
  1476. // the previous accept is probably from an old client with the same ID as this one;
  1477. // this is bad, disconnect client
  1478. client_log(client, BLOG_ERROR, "acceptpeer: already accepted to %d", (int)id);
  1479. client_remove(client);
  1480. return;
  1481. }
  1482. client_log(client, BLOG_INFO, "accepted %d", (int)id);
  1483. // set accepted
  1484. flow->accepted = 1;
  1485. // if pair is resetting, continue
  1486. if (flow->resetting) {
  1487. peer_flow_drive_reset(flow);
  1488. } else if (flow->opposite->resetting) {
  1489. peer_flow_drive_reset(flow->opposite);
  1490. }
  1491. }
  1492. struct peer_flow * peer_flow_create (struct client_data *src_client, struct client_data *dest_client)
  1493. {
  1494. ASSERT(src_client->initstatus == INITSTATUS_COMPLETE)
  1495. ASSERT(!src_client->dying)
  1496. ASSERT(dest_client->initstatus == INITSTATUS_COMPLETE)
  1497. ASSERT(!dest_client->dying)
  1498. ASSERT(!find_flow(src_client, dest_client->id))
  1499. // allocate flow structure
  1500. struct peer_flow *flow = (struct peer_flow *)malloc(sizeof(*flow));
  1501. if (!flow) {
  1502. BLog(BLOG_ERROR, "malloc failed");
  1503. goto fail0;
  1504. }
  1505. // set source and destination
  1506. flow->src_client = src_client;
  1507. flow->dest_client = dest_client;
  1508. flow->dest_client_id = dest_client->id;
  1509. // add to source list and tree
  1510. LinkedList1_Append(&flow->src_client->peer_out_flows_list, &flow->src_list_node);
  1511. ASSERT_EXECUTE(BAVL_Insert(&flow->src_client->peer_out_flows_tree, &flow->src_tree_node, NULL))
  1512. // add to destination client list
  1513. LinkedList1_Append(&flow->dest_client->output_peers_flows, &flow->dest_list_node);
  1514. // have no I/O
  1515. flow->have_io = 0;
  1516. // init reset timer
  1517. BTimer_Init(&flow->reset_timer, CLIENT_RESET_TIME, (BTimer_handler)peer_flow_reset_timer_handler, flow);
  1518. return flow;
  1519. fail0:
  1520. return NULL;
  1521. }
  1522. void peer_flow_dealloc (struct peer_flow *flow)
  1523. {
  1524. if (flow->have_io) { PacketPassFairQueueFlow_AssertFree(&flow->qflow); }
  1525. // free reset timer
  1526. BReactor_RemoveTimer(&ss, &flow->reset_timer);
  1527. // free I/O
  1528. if (flow->have_io) {
  1529. peer_flow_free_io(flow);
  1530. }
  1531. // remove from destination client list
  1532. LinkedList1_Remove(&flow->dest_client->output_peers_flows, &flow->dest_list_node);
  1533. // remove from source list and hash table
  1534. if (flow->src_client) {
  1535. BAVL_Remove(&flow->src_client->peer_out_flows_tree, &flow->src_tree_node);
  1536. LinkedList1_Remove(&flow->src_client->peer_out_flows_list, &flow->src_list_node);
  1537. }
  1538. // free memory
  1539. free(flow);
  1540. }
  1541. int peer_flow_init_io (struct peer_flow *flow)
  1542. {
  1543. ASSERT(!flow->have_io)
  1544. // init queue flow
  1545. PacketPassFairQueueFlow_Init(&flow->qflow, &flow->dest_client->output_peers_fairqueue);
  1546. // init PacketProtoFlow
  1547. if (!PacketProtoFlow_Init(
  1548. &flow->oflow, SC_MAX_ENC, CLIENT_PEER_FLOW_BUFFER_MIN_PACKETS,
  1549. PacketPassFairQueueFlow_GetInput(&flow->qflow), BReactor_PendingGroup(&ss)
  1550. )) {
  1551. BLog(BLOG_ERROR, "PacketProtoFlow_Init failed");
  1552. goto fail1;
  1553. }
  1554. flow->input = PacketProtoFlow_GetInput(&flow->oflow);
  1555. // set no packet
  1556. flow->packet_len = -1;
  1557. // set have I/O
  1558. flow->have_io = 1;
  1559. return 1;
  1560. fail1:
  1561. PacketPassFairQueueFlow_Free(&flow->qflow);
  1562. return 0;
  1563. }
  1564. void peer_flow_free_io (struct peer_flow *flow)
  1565. {
  1566. ASSERT(flow->have_io)
  1567. PacketPassFairQueueFlow_AssertFree(&flow->qflow);
  1568. // free PacketProtoFlow
  1569. PacketProtoFlow_Free(&flow->oflow);
  1570. // free queue flow
  1571. PacketPassFairQueueFlow_Free(&flow->qflow);
  1572. // set have no I/O
  1573. flow->have_io = 0;
  1574. }
  1575. void peer_flow_disconnect (struct peer_flow *flow)
  1576. {
  1577. ASSERT(flow->src_client)
  1578. ASSERT(flow->dest_client->initstatus == INITSTATUS_COMPLETE)
  1579. ASSERT(!flow->dest_client->dying)
  1580. ASSERT(flow->have_io)
  1581. ASSERT(PacketPassFairQueueFlow_IsBusy(&flow->qflow))
  1582. // stop reset timer
  1583. BReactor_RemoveTimer(&ss, &flow->reset_timer);
  1584. // remove from source list and hash table
  1585. BAVL_Remove(&flow->src_client->peer_out_flows_tree, &flow->src_tree_node);
  1586. LinkedList1_Remove(&flow->src_client->peer_out_flows_list, &flow->src_list_node);
  1587. // set no source
  1588. flow->src_client = NULL;
  1589. // set busy handler
  1590. PacketPassFairQueueFlow_SetBusyHandler(&flow->qflow, (PacketPassFairQueue_handler_busy)peer_flow_handler_canremove, flow);
  1591. }
  1592. int peer_flow_start_packet (struct peer_flow *flow, void **data, int len)
  1593. {
  1594. ASSERT(flow->dest_client->initstatus == INITSTATUS_COMPLETE)
  1595. ASSERT(!flow->dest_client->dying)
  1596. ASSERT(flow->src_client->initstatus == INITSTATUS_COMPLETE)
  1597. ASSERT(!flow->src_client->dying)
  1598. ASSERT(!flow->resetting)
  1599. ASSERT(!flow->opposite->resetting)
  1600. ASSERT(flow->have_io)
  1601. ASSERT(flow->packet_len == -1)
  1602. ASSERT(len >= 0)
  1603. ASSERT(len <= SC_MAX_PAYLOAD)
  1604. ASSERT(!(len > 0) || data)
  1605. // obtain location for writing the packet
  1606. if (!BufferWriter_StartPacket(flow->input, &flow->packet)) {
  1607. return 0;
  1608. }
  1609. // remember packet length
  1610. flow->packet_len = len;
  1611. if (data) {
  1612. *data = flow->packet + sizeof(struct sc_header);
  1613. }
  1614. return 1;
  1615. }
  1616. void peer_flow_end_packet (struct peer_flow *flow, uint8_t type)
  1617. {
  1618. ASSERT(flow->have_io)
  1619. ASSERT(flow->packet_len >= 0)
  1620. ASSERT(flow->packet_len <= SC_MAX_PAYLOAD)
  1621. // write header
  1622. struct sc_header header;
  1623. header.type = type;
  1624. memcpy(flow->packet, &header, sizeof(header));
  1625. // finish writing packet
  1626. BufferWriter_EndPacket(flow->input, sizeof(struct sc_header) + flow->packet_len);
  1627. // set have no packet
  1628. flow->packet_len = -1;
  1629. }
  1630. void peer_flow_handler_canremove (struct peer_flow *flow)
  1631. {
  1632. ASSERT(!flow->src_client)
  1633. ASSERT(flow->dest_client->initstatus == INITSTATUS_COMPLETE)
  1634. ASSERT(!flow->dest_client->dying)
  1635. ASSERT(flow->have_io)
  1636. PacketPassFairQueueFlow_AssertFree(&flow->qflow);
  1637. client_log(flow->dest_client, BLOG_DEBUG, "removing old flow");
  1638. peer_flow_dealloc(flow);
  1639. return;
  1640. }
  1641. void peer_flow_start_reset (struct peer_flow *flow)
  1642. {
  1643. ASSERT(flow->src_client->initstatus == INITSTATUS_COMPLETE)
  1644. ASSERT(!flow->src_client->dying)
  1645. ASSERT(flow->dest_client->initstatus == INITSTATUS_COMPLETE)
  1646. ASSERT(!flow->dest_client->dying)
  1647. ASSERT(!flow->resetting)
  1648. ASSERT(!flow->opposite->resetting)
  1649. ASSERT(flow->have_io)
  1650. ASSERT(flow->opposite->have_io)
  1651. client_log(flow->src_client, BLOG_INFO, "starting reset to %d", (int)flow->dest_client->id);
  1652. // set resetting
  1653. flow->resetting = 1;
  1654. peer_flow_drive_reset(flow);
  1655. }
  1656. void peer_flow_drive_reset (struct peer_flow *flow)
  1657. {
  1658. ASSERT(flow->src_client->initstatus == INITSTATUS_COMPLETE)
  1659. ASSERT(!flow->src_client->dying)
  1660. ASSERT(flow->dest_client->initstatus == INITSTATUS_COMPLETE)
  1661. ASSERT(!flow->dest_client->dying)
  1662. ASSERT(flow->resetting)
  1663. ASSERT(!flow->opposite->resetting)
  1664. ASSERT(!BTimer_IsRunning(&flow->reset_timer))
  1665. // try to free I/O
  1666. if (flow->have_io) {
  1667. if (PacketPassFairQueueFlow_IsBusy(&flow->qflow)) {
  1668. PacketPassFairQueueFlow_SetBusyHandler(&flow->qflow, (PacketPassFairQueue_handler_busy)peer_flow_reset_qflow_handler_busy, flow);
  1669. } else {
  1670. peer_flow_free_io(flow);
  1671. }
  1672. }
  1673. // try to free opposite I/O
  1674. if (flow->opposite->have_io) {
  1675. if (PacketPassFairQueueFlow_IsBusy(&flow->opposite->qflow)) {
  1676. PacketPassFairQueueFlow_SetBusyHandler(&flow->opposite->qflow, (PacketPassFairQueue_handler_busy)peer_flow_reset_qflow_handler_busy, flow->opposite);
  1677. } else {
  1678. peer_flow_free_io(flow->opposite);
  1679. }
  1680. }
  1681. // if we still got some I/O, or some client hasn't accepted yet, wait
  1682. if (flow->have_io || flow->opposite->have_io || !flow->accepted || !flow->opposite->accepted) {
  1683. return;
  1684. }
  1685. // set reset timer
  1686. BReactor_SetTimer(&ss, &flow->reset_timer);
  1687. }
  1688. void peer_flow_reset_qflow_handler_busy (struct peer_flow *flow)
  1689. {
  1690. ASSERT(flow->src_client->initstatus == INITSTATUS_COMPLETE)
  1691. ASSERT(!flow->src_client->dying)
  1692. ASSERT(flow->dest_client->initstatus == INITSTATUS_COMPLETE)
  1693. ASSERT(!flow->dest_client->dying)
  1694. ASSERT(flow->resetting || flow->opposite->resetting)
  1695. ASSERT(flow->have_io)
  1696. ASSERT(!PacketPassFairQueueFlow_IsBusy(&flow->qflow))
  1697. if (flow->resetting) {
  1698. peer_flow_drive_reset(flow);
  1699. } else {
  1700. peer_flow_drive_reset(flow->opposite);
  1701. }
  1702. }
  1703. void peer_flow_reset_timer_handler (struct peer_flow *flow)
  1704. {
  1705. ASSERT(flow->src_client->initstatus == INITSTATUS_COMPLETE)
  1706. ASSERT(!flow->src_client->dying)
  1707. ASSERT(flow->dest_client->initstatus == INITSTATUS_COMPLETE)
  1708. ASSERT(!flow->dest_client->dying)
  1709. ASSERT(flow->resetting)
  1710. ASSERT(!flow->opposite->resetting)
  1711. ASSERT(!flow->have_io)
  1712. ASSERT(!flow->opposite->have_io)
  1713. ASSERT(flow->accepted)
  1714. ASSERT(flow->opposite->accepted)
  1715. client_log(flow->src_client, BLOG_INFO, "finally resetting to %d", (int)flow->dest_client->id);
  1716. struct peer_know *know = flow->know;
  1717. struct peer_know *know_opposite = flow->opposite->know;
  1718. // launch pair
  1719. if (!launch_pair(flow)) {
  1720. return;
  1721. }
  1722. // remove old knows
  1723. uninform_know(know);
  1724. uninform_know(know_opposite);
  1725. }
  1726. peerid_t new_client_id (void)
  1727. {
  1728. ASSERT(clients_num < options.max_clients)
  1729. for (int i = 0; i < options.max_clients; i++) {
  1730. peerid_t id = clients_nextid++;
  1731. if (!find_client_by_id(id)) {
  1732. return id;
  1733. }
  1734. }
  1735. ASSERT(0)
  1736. return 42;
  1737. }
  1738. struct client_data * find_client_by_id (peerid_t id)
  1739. {
  1740. BAVLNode *node;
  1741. if (!(node = BAVL_LookupExact(&clients_tree, &id))) {
  1742. return NULL;
  1743. }
  1744. return UPPER_OBJECT(node, struct client_data, tree_node);
  1745. }
  1746. int clients_allowed (struct client_data *client1, struct client_data *client2)
  1747. {
  1748. ASSERT(client1->initstatus == INITSTATUS_COMPLETE)
  1749. ASSERT(!client1->dying)
  1750. ASSERT(client2->initstatus == INITSTATUS_COMPLETE)
  1751. ASSERT(!client2->dying)
  1752. if (!options.comm_predicate) {
  1753. return 1;
  1754. }
  1755. // set values to compare against
  1756. comm_predicate_p1name = (client1->common_name ? client1->common_name : "");
  1757. comm_predicate_p2name = (client2->common_name ? client2->common_name : "");
  1758. BAddr_GetIPAddr(&client1->addr, &comm_predicate_p1addr);
  1759. BAddr_GetIPAddr(&client2->addr, &comm_predicate_p2addr);
  1760. // evaluate predicate
  1761. int res = BPredicate_Eval(&comm_predicate);
  1762. if (res < 0) {
  1763. return 0;
  1764. }
  1765. return res;
  1766. }
  1767. int comm_predicate_func_p1name_cb (void *user, void **args)
  1768. {
  1769. char *arg = (char *)args[0];
  1770. return (!strcmp(arg, comm_predicate_p1name));
  1771. }
  1772. int comm_predicate_func_p2name_cb (void *user, void **args)
  1773. {
  1774. char *arg = (char *)args[0];
  1775. return (!strcmp(arg, comm_predicate_p2name));
  1776. }
  1777. int comm_predicate_func_p1addr_cb (void *user, void **args)
  1778. {
  1779. char *arg = (char *)args[0];
  1780. BIPAddr addr;
  1781. if (!BIPAddr_Resolve(&addr, arg, 1)) {
  1782. BLog(BLOG_WARNING, "failed to parse address");
  1783. return -1;
  1784. }
  1785. return BIPAddr_Compare(&addr, &comm_predicate_p1addr);
  1786. }
  1787. int comm_predicate_func_p2addr_cb (void *user, void **args)
  1788. {
  1789. char *arg = (char *)args[0];
  1790. BIPAddr addr;
  1791. if (!BIPAddr_Resolve(&addr, arg, 1)) {
  1792. BLog(BLOG_WARNING, "failed to parse address");
  1793. return -1;
  1794. }
  1795. return BIPAddr_Compare(&addr, &comm_predicate_p2addr);
  1796. }
  1797. int relay_allowed (struct client_data *client, struct client_data *relay)
  1798. {
  1799. if (!options.relay_predicate) {
  1800. return 0;
  1801. }
  1802. // set values to compare against
  1803. relay_predicate_pname = (client->common_name ? client->common_name : "");
  1804. relay_predicate_rname = (relay->common_name ? relay->common_name : "");
  1805. BAddr_GetIPAddr(&client->addr, &relay_predicate_paddr);
  1806. BAddr_GetIPAddr(&relay->addr, &relay_predicate_raddr);
  1807. // evaluate predicate
  1808. int res = BPredicate_Eval(&relay_predicate);
  1809. if (res < 0) {
  1810. return 0;
  1811. }
  1812. return res;
  1813. }
  1814. int relay_predicate_func_pname_cb (void *user, void **args)
  1815. {
  1816. char *arg = (char *)args[0];
  1817. return (!strcmp(arg, relay_predicate_pname));
  1818. }
  1819. int relay_predicate_func_rname_cb (void *user, void **args)
  1820. {
  1821. char *arg = (char *)args[0];
  1822. return (!strcmp(arg, relay_predicate_rname));
  1823. }
  1824. int relay_predicate_func_paddr_cb (void *user, void **args)
  1825. {
  1826. char *arg = (char *)args[0];
  1827. BIPAddr addr;
  1828. if (!BIPAddr_Resolve(&addr, arg, 1)) {
  1829. BLog(BLOG_ERROR, "paddr: failed to parse address");
  1830. return -1;
  1831. }
  1832. return BIPAddr_Compare(&addr, &relay_predicate_paddr);
  1833. }
  1834. int relay_predicate_func_raddr_cb (void *user, void **args)
  1835. {
  1836. char *arg = (char *)args[0];
  1837. BIPAddr addr;
  1838. if (!BIPAddr_Resolve(&addr, arg, 1)) {
  1839. BLog(BLOG_ERROR, "raddr: failed to parse address");
  1840. return -1;
  1841. }
  1842. return BIPAddr_Compare(&addr, &relay_predicate_raddr);
  1843. }
  1844. int peerid_comparator (void *unused, peerid_t *p1, peerid_t *p2)
  1845. {
  1846. return B_COMPARE(*p1, *p2);
  1847. }
  1848. struct peer_know * create_know (struct client_data *from, struct client_data *to, int relay_server, int relay_client)
  1849. {
  1850. ASSERT(from->initstatus == INITSTATUS_COMPLETE)
  1851. ASSERT(!from->dying)
  1852. ASSERT(to->initstatus == INITSTATUS_COMPLETE)
  1853. ASSERT(!to->dying)
  1854. // allocate structure
  1855. struct peer_know *k = (struct peer_know *)malloc(sizeof(*k));
  1856. if (!k) {
  1857. return NULL;
  1858. }
  1859. // init arguments
  1860. k->from = from;
  1861. k->to = to;
  1862. k->relay_server = relay_server;
  1863. k->relay_client = relay_client;
  1864. // append to lists
  1865. LinkedList1_Append(&from->know_out_list, &k->from_node);
  1866. LinkedList1_Append(&to->know_in_list, &k->to_node);
  1867. // init and set inform job to inform client 'from' about client 'to'
  1868. BPending_Init(&k->inform_job, BReactor_PendingGroup(&ss), (BPending_handler)know_inform_job_handler, k);
  1869. BPending_Set(&k->inform_job);
  1870. // init uninform job
  1871. BPending_Init(&k->uninform_job, BReactor_PendingGroup(&ss), (BPending_handler)know_uninform_job_handler, k);
  1872. return k;
  1873. }
  1874. void remove_know (struct peer_know *k)
  1875. {
  1876. // free uninform job
  1877. BPending_Free(&k->uninform_job);
  1878. // free inform job
  1879. BPending_Free(&k->inform_job);
  1880. // remove from lists
  1881. LinkedList1_Remove(&k->to->know_in_list, &k->to_node);
  1882. LinkedList1_Remove(&k->from->know_out_list, &k->from_node);
  1883. // free structure
  1884. free(k);
  1885. }
  1886. void know_inform_job_handler (struct peer_know *k)
  1887. {
  1888. ASSERT(!k->from->dying)
  1889. ASSERT(!k->to->dying)
  1890. client_send_newclient(k->from, k->to, k->relay_server, k->relay_client);
  1891. return;
  1892. }
  1893. void uninform_know (struct peer_know *k)
  1894. {
  1895. ASSERT(!k->from->dying)
  1896. // if 'from' has not been informed about 'to' yet, remove know, otherwise
  1897. // schedule informing 'from' that 'to' is no more
  1898. if (BPending_IsSet(&k->inform_job)) {
  1899. remove_know(k);
  1900. } else {
  1901. BPending_Set(&k->uninform_job);
  1902. }
  1903. }
  1904. void know_uninform_job_handler (struct peer_know *k)
  1905. {
  1906. ASSERT(!k->from->dying)
  1907. ASSERT(!BPending_IsSet(&k->inform_job))
  1908. struct client_data *from = k->from;
  1909. struct client_data *to = k->to;
  1910. // remove know
  1911. remove_know(k);
  1912. // uninform
  1913. client_send_endclient(from, to->id);
  1914. }
  1915. int launch_pair (struct peer_flow *flow_to)
  1916. {
  1917. struct client_data *client = flow_to->src_client;
  1918. struct client_data *client2 = flow_to->dest_client;
  1919. ASSERT(client->initstatus == INITSTATUS_COMPLETE)
  1920. ASSERT(!client->dying)
  1921. ASSERT(client2->initstatus == INITSTATUS_COMPLETE)
  1922. ASSERT(!client2->dying)
  1923. ASSERT(!flow_to->have_io)
  1924. ASSERT(!flow_to->opposite->have_io)
  1925. ASSERT(!BTimer_IsRunning(&flow_to->reset_timer))
  1926. ASSERT(!BTimer_IsRunning(&flow_to->opposite->reset_timer))
  1927. // init I/O
  1928. if (!peer_flow_init_io(flow_to)) {
  1929. goto fail;
  1930. }
  1931. // init opposite I/O
  1932. if (!peer_flow_init_io(flow_to->opposite)) {
  1933. goto fail;
  1934. }
  1935. // determine relay relations
  1936. int relay_to = relay_allowed(client, client2);
  1937. int relay_from = relay_allowed(client2, client);
  1938. // create know to
  1939. struct peer_know *know_to = create_know(client, client2, relay_to, relay_from);
  1940. if (!know_to) {
  1941. client_log(client, BLOG_ERROR, "failed to allocate know to %d", (int)client2->id);
  1942. goto fail;
  1943. }
  1944. // create know from
  1945. struct peer_know *know_from = create_know(client2, client, relay_from, relay_to);
  1946. if (!know_from) {
  1947. client_log(client, BLOG_ERROR, "failed to allocate know from %d", (int)client2->id);
  1948. goto fail;
  1949. }
  1950. // set know pointers in flows
  1951. flow_to->know = know_to;
  1952. flow_to->opposite->know = know_from;
  1953. // set not accepted, or assume accepted for old version
  1954. flow_to->accepted = (flow_to->src_client->version <= SC_OLDVERSION_NOSSL);
  1955. flow_to->opposite->accepted = (flow_to->opposite->src_client->version <= SC_OLDVERSION_NOSSL);
  1956. // set not resetting
  1957. flow_to->resetting = 0;
  1958. flow_to->opposite->resetting = 0;
  1959. return 1;
  1960. fail:
  1961. client_remove(client);
  1962. return 0;
  1963. }
  1964. struct peer_flow * find_flow (struct client_data *client, peerid_t dest_id)
  1965. {
  1966. ASSERT(client->initstatus == INITSTATUS_COMPLETE)
  1967. ASSERT(!client->dying)
  1968. BAVLNode *node = BAVL_LookupExact(&client->peer_out_flows_tree, &dest_id);
  1969. if (!node) {
  1970. return NULL;
  1971. }
  1972. struct peer_flow *flow = UPPER_OBJECT(node, struct peer_flow, src_tree_node);
  1973. ASSERT(flow->dest_client->id == dest_id)
  1974. ASSERT(flow->dest_client->initstatus == INITSTATUS_COMPLETE)
  1975. ASSERT(!flow->dest_client->dying)
  1976. return flow;
  1977. }