You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

rspamd.c 41KB

10 years ago
16 years ago
16 years ago
9 years ago
9 years ago
9 years ago
9 years ago
12 years ago
12 years ago
12 years ago
12 years ago
12 years ago
12 years ago
16 years ago
16 years ago
16 years ago
16 years ago
16 years ago
16 years ago
16 years ago
16 years ago
16 years ago
16 years ago
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011101210131014101510161017101810191020102110221023102410251026102710281029103010311032103310341035103610371038103910401041104210431044104510461047104810491050105110521053105410551056105710581059106010611062106310641065106610671068106910701071107210731074107510761077107810791080108110821083108410851086108710881089109010911092109310941095109610971098109911001101110211031104110511061107110811091110111111121113111411151116111711181119112011211122112311241125112611271128112911301131113211331134113511361137113811391140114111421143114411451146114711481149115011511152115311541155115611571158115911601161116211631164116511661167116811691170117111721173117411751176117711781179118011811182118311841185118611871188118911901191119211931194119511961197119811991200120112021203120412051206120712081209121012111212121312141215121612171218121912201221122212231224122512261227122812291230123112321233123412351236123712381239124012411242124312441245124612471248124912501251125212531254125512561257125812591260126112621263126412651266126712681269127012711272127312741275127612771278127912801281128212831284128512861287128812891290129112921293129412951296129712981299130013011302130313041305130613071308130913101311131213131314131513161317131813191320132113221323132413251326132713281329133013311332133313341335133613371338133913401341134213431344134513461347134813491350135113521353135413551356135713581359136013611362136313641365136613671368136913701371137213731374137513761377137813791380138113821383138413851386138713881389139013911392139313941395139613971398139914001401140214031404140514061407140814091410141114121413141414151416141714181419142014211422142314241425142614271428142914301431143214331434143514361437143814391440144114421443144414451446144714481449145014511452145314541455145614571458145914601461146214631464146514661467146814691470147114721473147414751476147714781479148014811482148314841485148614871488148914901491149214931494149514961497149814991500150115021503150415051506150715081509151015111512151315141515151615171518151915201521152215231524152515261527152815291530153115321533153415351536153715381539154015411542154315441545154615471548154915501551155215531554155515561557155815591560156115621563156415651566156715681569
  1. /*-
  2. * Copyright 2016 Vsevolod Stakhov
  3. *
  4. * Licensed under the Apache License, Version 2.0 (the "License");
  5. * you may not use this file except in compliance with the License.
  6. * You may obtain a copy of the License at
  7. *
  8. * http://www.apache.org/licenses/LICENSE-2.0
  9. *
  10. * Unless required by applicable law or agreed to in writing, software
  11. * distributed under the License is distributed on an "AS IS" BASIS,
  12. * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13. * See the License for the specific language governing permissions and
  14. * limitations under the License.
  15. */
  16. #include "config.h"
  17. #include "rspamd.h"
  18. #include "libserver/maps/map.h"
  19. #include "lua/lua_common.h"
  20. #include "libserver/worker_util.h"
  21. #include "libserver/rspamd_control.h"
  22. #include "ottery.h"
  23. #include "cryptobox.h"
  24. #include "utlist.h"
  25. #include "unix-std.h"
  26. /* pwd and grp */
  27. #ifdef HAVE_PWD_H
  28. #include <pwd.h>
  29. #endif
  30. #ifdef HAVE_GRP_H
  31. #include <grp.h>
  32. #endif
  33. #ifdef HAVE_NFTW
  34. #include <ftw.h>
  35. #endif
  36. #include <signal.h>
  37. #ifdef HAVE_SYS_RESOURCE_H
  38. #include <sys/resource.h>
  39. #endif
  40. #ifdef HAVE_LIBUTIL_H
  41. #include <libutil.h>
  42. #endif
  43. #ifdef HAVE_OPENSSL
  44. #include <openssl/err.h>
  45. #include <openssl/evp.h>
  46. #endif
  47. #include "sqlite3.h"
  48. #include "contrib/libev/ev.h"
  49. /* 2 seconds to fork new process in place of dead one */
  50. #define SOFT_FORK_TIME 2
  51. /* 10 seconds after getting termination signal to terminate all workers with SIGKILL */
  52. #define TERMINATION_INTERVAL (0.2)
  53. static gboolean load_rspamd_config (struct rspamd_main *rspamd_main,
  54. struct rspamd_config *cfg,
  55. gboolean init_modules,
  56. enum rspamd_post_load_options opts,
  57. gboolean reload);
  58. static void rspamd_cld_handler (EV_P_ ev_child *w,
  59. struct rspamd_main *rspamd_main,
  60. struct rspamd_worker *wrk);
  61. /* Control socket */
  62. static gint control_fd;
  63. static ev_io control_ev;
  64. static struct rspamd_stat old_stat;
  65. static ev_timer stat_ev;
  66. static gboolean valgrind_mode = FALSE;
  67. /* Cmdline options */
  68. static gboolean no_fork = FALSE;
  69. static gboolean show_version = FALSE;
  70. static gchar **cfg_names = NULL;
  71. static gchar *rspamd_user = NULL;
  72. static gchar *rspamd_group = NULL;
  73. static gchar *rspamd_pidfile = NULL;
  74. static gboolean is_debug = FALSE;
  75. static gboolean is_insecure = FALSE;
  76. static GHashTable *ucl_vars = NULL;
  77. static gchar **lua_env = NULL;
  78. static gboolean skip_template = FALSE;
  79. static gint term_attempts = 0;
  80. /* List of active listen sockets indexed by worker type */
  81. static GHashTable *listen_sockets = NULL;
  82. /* Defined in modules.c */
  83. extern module_t *modules[];
  84. extern worker_t *workers[];
  85. /* Command line options */
  86. static gboolean rspamd_parse_var (const gchar *option_name,
  87. const gchar *value, gpointer data,
  88. GError **error);
  89. static GOptionEntry entries[] =
  90. {
  91. { "no-fork", 'f', 0, G_OPTION_ARG_NONE, &no_fork,
  92. "Do not daemonize main process", NULL },
  93. { "config", 'c', 0, G_OPTION_ARG_FILENAME_ARRAY, &cfg_names,
  94. "Specify config file(s)", NULL },
  95. { "user", 'u', 0, G_OPTION_ARG_STRING, &rspamd_user,
  96. "User to run rspamd as", NULL },
  97. { "group", 'g', 0, G_OPTION_ARG_STRING, &rspamd_group,
  98. "Group to run rspamd as", NULL },
  99. { "pid", 'p', 0, G_OPTION_ARG_STRING, &rspamd_pidfile, "Path to pidfile",
  100. NULL },
  101. { "debug", 'd', 0, G_OPTION_ARG_NONE, &is_debug, "Force debug output",
  102. NULL },
  103. { "insecure", 'i', 0, G_OPTION_ARG_NONE, &is_insecure,
  104. "Ignore running workers as privileged users (insecure)", NULL },
  105. { "version", 'v', 0, G_OPTION_ARG_NONE, &show_version,
  106. "Show version and exit", NULL },
  107. {"var", 0, 0, G_OPTION_ARG_CALLBACK, (gpointer)&rspamd_parse_var,
  108. "Redefine/define environment variable", NULL},
  109. {"skip-template", 'T', 0, G_OPTION_ARG_NONE, &skip_template,
  110. "Do not apply Jinja templates", NULL},
  111. {"lua-env", '\0', 0, G_OPTION_ARG_FILENAME_ARRAY, &lua_env,
  112. "Load lua environment from the specified files", NULL},
  113. { NULL, 0, 0, G_OPTION_ARG_NONE, NULL, NULL, NULL }
  114. };
  115. static gboolean
  116. rspamd_parse_var (const gchar *option_name,
  117. const gchar *value, gpointer data,
  118. GError **error)
  119. {
  120. gchar *k, *v, *t;
  121. t = strchr (value, '=');
  122. if (t != NULL) {
  123. k = g_strdup (value);
  124. t = k + (t - value);
  125. v = g_strdup (t + 1);
  126. *t = '\0';
  127. if (ucl_vars == NULL) {
  128. ucl_vars = g_hash_table_new_full (rspamd_strcase_hash,
  129. rspamd_strcase_equal, g_free, g_free);
  130. }
  131. g_hash_table_insert (ucl_vars, k, v);
  132. }
  133. else {
  134. g_set_error (error, g_quark_try_string ("main"), EINVAL,
  135. "Bad variable format: %s", value);
  136. return FALSE;
  137. }
  138. return TRUE;
  139. }
  140. static void
  141. read_cmd_line (gint *argc, gchar ***argv, struct rspamd_config *cfg)
  142. {
  143. GError *error = NULL;
  144. GOptionContext *context;
  145. guint cfg_num;
  146. context = g_option_context_new ("- run rspamd daemon");
  147. #if defined(GIT_VERSION) && GIT_VERSION == 1
  148. g_option_context_set_summary (context,
  149. "Summary:\n Rspamd daemon version " RVERSION "-git\n Git id: " RID);
  150. #else
  151. g_option_context_set_summary (context,
  152. "Summary:\n Rspamd daemon version " RVERSION);
  153. #endif
  154. g_option_context_add_main_entries (context, entries, NULL);
  155. if (!g_option_context_parse (context, argc, argv, &error)) {
  156. fprintf (stderr, "option parsing failed: %s\n", error->message);
  157. g_option_context_free (context);
  158. exit (1);
  159. }
  160. cfg->rspamd_user = rspamd_user;
  161. cfg->rspamd_group = rspamd_group;
  162. cfg_num = cfg_names != NULL ? g_strv_length (cfg_names) : 0;
  163. if (cfg_num == 0) {
  164. cfg->cfg_name = FIXED_CONFIG_FILE;
  165. }
  166. else {
  167. cfg->cfg_name = cfg_names[0];
  168. g_assert (cfg_num == 1);
  169. }
  170. cfg->pid_file = rspamd_pidfile;
  171. g_option_context_free (context);
  172. }
  173. static int
  174. rspamd_write_pid (struct rspamd_main *main)
  175. {
  176. pid_t pid;
  177. if (main->cfg->pid_file == NULL) {
  178. return -1;
  179. }
  180. main->pfh = rspamd_pidfile_open (main->cfg->pid_file, 0644, &pid);
  181. if (main->pfh == NULL) {
  182. return -1;
  183. }
  184. if (main->is_privilleged) {
  185. /* Force root user as owner of pid file */
  186. #ifdef HAVE_PIDFILE_FILENO
  187. if (fchown (pidfile_fileno (main->pfh), 0, 0) == -1) {
  188. #else
  189. if (fchown (main->pfh->pf_fd, 0, 0) == -1) {
  190. #endif
  191. }
  192. }
  193. rspamd_pidfile_write (main->pfh);
  194. return 0;
  195. }
  196. /* Detect privilleged mode */
  197. static void
  198. detect_priv (struct rspamd_main *rspamd_main)
  199. {
  200. struct passwd *pwd;
  201. struct group *grp;
  202. uid_t euid;
  203. euid = geteuid ();
  204. if (euid == 0) {
  205. if (!rspamd_main->cfg->rspamd_user && !is_insecure) {
  206. msg_err_main (
  207. "cannot run rspamd workers as root user, please add -u and -g options to select a proper unprivilleged user or specify --insecure flag");
  208. exit (EXIT_FAILURE);
  209. }
  210. else if (is_insecure) {
  211. rspamd_main->is_privilleged = TRUE;
  212. rspamd_main->workers_uid = 0;
  213. rspamd_main->workers_gid = 0;
  214. }
  215. else {
  216. rspamd_main->is_privilleged = TRUE;
  217. pwd = getpwnam (rspamd_main->cfg->rspamd_user);
  218. if (pwd == NULL) {
  219. msg_err_main ("user specified does not exists (%s), aborting",
  220. strerror (errno));
  221. exit (-errno);
  222. }
  223. if (rspamd_main->cfg->rspamd_group) {
  224. grp = getgrnam (rspamd_main->cfg->rspamd_group);
  225. if (grp == NULL) {
  226. msg_err_main ("group specified does not exists (%s), aborting",
  227. strerror (errno));
  228. exit (-errno);
  229. }
  230. rspamd_main->workers_gid = grp->gr_gid;
  231. }
  232. else {
  233. rspamd_main->workers_gid = (gid_t)-1;
  234. }
  235. rspamd_main->workers_uid = pwd->pw_uid;
  236. }
  237. }
  238. else {
  239. rspamd_main->is_privilleged = FALSE;
  240. rspamd_main->workers_uid = (uid_t)-1;
  241. rspamd_main->workers_gid = (gid_t)-1;
  242. }
  243. }
  244. static void
  245. config_logger (rspamd_mempool_t *pool, gpointer ud)
  246. {
  247. struct rspamd_main *rspamd_main = ud;
  248. rspamd_main->logger = rspamd_log_open_specific (rspamd_main->server_pool,
  249. rspamd_main->cfg,
  250. "main",
  251. rspamd_main->workers_uid,
  252. rspamd_main->workers_gid);
  253. if (rspamd_main->logger == NULL) {
  254. /*
  255. * XXX:
  256. * Error has been already logged (in fact,
  257. * we might fall back to console logger here)
  258. */
  259. exit (EXIT_FAILURE);
  260. }
  261. rspamd_logger_configure_modules (rspamd_main->cfg->debug_modules);
  262. }
  263. static gboolean
  264. reread_config (struct rspamd_main *rspamd_main)
  265. {
  266. struct rspamd_config *tmp_cfg, *old_cfg;
  267. gchar *cfg_file;
  268. int load_opts = RSPAMD_CONFIG_INIT_VALIDATE|RSPAMD_CONFIG_INIT_SYMCACHE|
  269. RSPAMD_CONFIG_INIT_LIBS|RSPAMD_CONFIG_INIT_URL;
  270. rspamd_symcache_save (rspamd_main->cfg->cache);
  271. tmp_cfg = rspamd_config_new (RSPAMD_CONFIG_INIT_DEFAULT);
  272. tmp_cfg->libs_ctx = rspamd_main->cfg->libs_ctx;
  273. REF_RETAIN (tmp_cfg->libs_ctx);
  274. cfg_file = rspamd_mempool_strdup (tmp_cfg->cfg_pool,
  275. rspamd_main->cfg->cfg_name);
  276. /* Save some variables */
  277. tmp_cfg->cfg_name = cfg_file;
  278. old_cfg = rspamd_main->cfg;
  279. rspamd_main->cfg = tmp_cfg;
  280. rspamd_logger_t *old_logger = rspamd_main->logger;
  281. if (!load_rspamd_config (rspamd_main, tmp_cfg, TRUE, load_opts, TRUE)) {
  282. rspamd_main->cfg = old_cfg;
  283. rspamd_main->logger = old_logger;
  284. msg_err_main ("cannot parse new config file, revert to old one");
  285. REF_RELEASE (tmp_cfg);
  286. return FALSE;
  287. }
  288. else {
  289. rspamd_log_close (old_logger);
  290. msg_info_main ("replacing config");
  291. REF_RELEASE (old_cfg);
  292. rspamd_main->cfg->rspamd_user = rspamd_user;
  293. rspamd_main->cfg->rspamd_group = rspamd_group;
  294. /* Here, we can do post actions with the existing config */
  295. /*
  296. * As some rules are defined in lua, we need to process them, then init
  297. * modules and merely afterwards to init modules
  298. */
  299. rspamd_lua_post_load_config (tmp_cfg);
  300. rspamd_init_filters (tmp_cfg, true, false);
  301. /* Do post-load actions */
  302. rspamd_config_post_load (tmp_cfg,
  303. load_opts|RSPAMD_CONFIG_INIT_POST_LOAD_LUA|RSPAMD_CONFIG_INIT_PRELOAD_MAPS);
  304. msg_info_main ("config has been reread successfully");
  305. }
  306. return TRUE;
  307. }
  308. struct waiting_worker {
  309. struct rspamd_main *rspamd_main;
  310. struct ev_timer wait_ev;
  311. struct rspamd_worker_conf *cf;
  312. guint oldindex;
  313. };
  314. static void
  315. rspamd_fork_delayed_cb (EV_P_ ev_timer *w, int revents)
  316. {
  317. struct waiting_worker *waiting_worker = (struct waiting_worker *)w->data;
  318. ev_timer_stop (EV_A_ &waiting_worker->wait_ev);
  319. rspamd_fork_worker (waiting_worker->rspamd_main, waiting_worker->cf,
  320. waiting_worker->oldindex,
  321. waiting_worker->rspamd_main->event_loop,
  322. rspamd_cld_handler, listen_sockets);
  323. REF_RELEASE (waiting_worker->cf);
  324. g_free (waiting_worker);
  325. }
  326. static void
  327. rspamd_fork_delayed (struct rspamd_worker_conf *cf,
  328. guint index,
  329. struct rspamd_main *rspamd_main)
  330. {
  331. struct waiting_worker *nw;
  332. nw = g_malloc0 (sizeof (*nw));
  333. nw->cf = cf;
  334. nw->oldindex = index;
  335. nw->rspamd_main = rspamd_main;
  336. REF_RETAIN (cf);
  337. nw->wait_ev.data = nw;
  338. ev_timer_init (&nw->wait_ev, rspamd_fork_delayed_cb, SOFT_FORK_TIME, 0.0);
  339. ev_timer_start (rspamd_main->event_loop, &nw->wait_ev);
  340. }
  341. static GList *
  342. create_listen_socket (GPtrArray *addrs, guint cnt,
  343. enum rspamd_worker_socket_type listen_type)
  344. {
  345. GList *result = NULL;
  346. gint fd;
  347. guint i;
  348. static const int listen_opts = RSPAMD_INET_ADDRESS_LISTEN_ASYNC;
  349. struct rspamd_worker_listen_socket *ls;
  350. g_ptr_array_sort (addrs, rspamd_inet_address_compare_ptr);
  351. for (i = 0; i < cnt; i ++) {
  352. /*
  353. * Copy address to avoid reload issues
  354. */
  355. if (listen_type & RSPAMD_WORKER_SOCKET_TCP) {
  356. fd = rspamd_inet_address_listen (g_ptr_array_index (addrs, i),
  357. SOCK_STREAM,
  358. listen_opts, -1);
  359. if (fd != -1) {
  360. ls = g_malloc0 (sizeof (*ls));
  361. ls->addr = rspamd_inet_address_copy (g_ptr_array_index (addrs, i));
  362. ls->fd = fd;
  363. ls->type = RSPAMD_WORKER_SOCKET_TCP;
  364. result = g_list_prepend (result, ls);
  365. }
  366. }
  367. if (listen_type & RSPAMD_WORKER_SOCKET_UDP) {
  368. fd = rspamd_inet_address_listen (g_ptr_array_index (addrs, i),
  369. SOCK_DGRAM,
  370. listen_opts | RSPAMD_INET_ADDRESS_LISTEN_REUSEPORT, -1);
  371. if (fd != -1) {
  372. ls = g_malloc0 (sizeof (*ls));
  373. ls->addr = rspamd_inet_address_copy (g_ptr_array_index (addrs, i));
  374. ls->fd = fd;
  375. ls->type = RSPAMD_WORKER_SOCKET_UDP;
  376. result = g_list_prepend (result, ls);
  377. }
  378. }
  379. }
  380. return result;
  381. }
  382. static GList *
  383. systemd_get_socket (struct rspamd_main *rspamd_main, const gchar *fdname)
  384. {
  385. int number, sock, num_passed, flags;
  386. GList *result = NULL;
  387. const gchar *e;
  388. gchar **fdnames;
  389. gchar *end;
  390. struct stat st;
  391. static const int sd_listen_fds_start = 3; /* SD_LISTEN_FDS_START */
  392. struct rspamd_worker_listen_socket *ls;
  393. union {
  394. struct sockaddr_storage ss;
  395. struct sockaddr sa;
  396. } addr_storage;
  397. socklen_t slen = sizeof (addr_storage);
  398. gint stype;
  399. number = strtoul (fdname, &end, 10);
  400. if (end != NULL && *end != '\0') {
  401. /* Cannot parse as number, assume a name in LISTEN_FDNAMES. */
  402. e = getenv ("LISTEN_FDNAMES");
  403. if (!e) {
  404. msg_err_main ("cannot get systemd variable 'LISTEN_FDNAMES'");
  405. errno = ENOENT;
  406. return NULL;
  407. }
  408. fdnames = g_strsplit (e, ":", -1);
  409. for (number = 0; fdnames[number]; number++) {
  410. if (!strcmp (fdnames[number], fdname)) {
  411. break;
  412. }
  413. }
  414. if (!fdnames[number]) {
  415. number = -1;
  416. }
  417. g_strfreev (fdnames);
  418. }
  419. if (number < 0) {
  420. msg_warn_main ("cannot find systemd socket: %s", fdname);
  421. errno = ENOENT;
  422. return NULL;
  423. }
  424. e = getenv ("LISTEN_FDS");
  425. if (e != NULL) {
  426. errno = 0;
  427. num_passed = strtoul (e, &end, 10);
  428. if ((end == NULL || *end == '\0') && num_passed > number) {
  429. sock = number + sd_listen_fds_start;
  430. if (fstat (sock, &st) == -1) {
  431. msg_warn_main ("cannot stat systemd descriptor %d", sock);
  432. return NULL;
  433. }
  434. if (!S_ISSOCK (st.st_mode)) {
  435. msg_warn_main ("systemd descriptor %d is not a socket", sock);
  436. errno = EINVAL;
  437. return NULL;
  438. }
  439. flags = fcntl (sock, F_GETFD);
  440. if (flags != -1) {
  441. (void)fcntl (sock, F_SETFD, flags | FD_CLOEXEC);
  442. }
  443. rspamd_socket_nonblocking (sock);
  444. if (getsockname (sock, &addr_storage.sa, &slen) == -1) {
  445. msg_warn_main ("cannot get name for systemd descriptor %d: %s",
  446. sock, strerror (errno));
  447. errno = EINVAL;
  448. return NULL;
  449. }
  450. ls = g_malloc0 (sizeof (*ls));
  451. ls->addr = rspamd_inet_address_from_sa (&addr_storage.sa, slen);
  452. ls->fd = sock;
  453. ls->is_systemd = true;
  454. slen = sizeof (stype);
  455. if (getsockopt (sock, SOL_SOCKET, SO_TYPE, &stype, &slen) != -1) {
  456. if (stype == SOCK_STREAM) {
  457. ls->type = RSPAMD_WORKER_SOCKET_TCP;
  458. }
  459. else {
  460. ls->type = RSPAMD_WORKER_SOCKET_UDP;
  461. }
  462. }
  463. else {
  464. msg_warn_main ("cannot get type for systemd descriptor %d: %s",
  465. sock, strerror (errno));
  466. ls->type = RSPAMD_WORKER_SOCKET_TCP;
  467. }
  468. result = g_list_prepend (result, ls);
  469. }
  470. else if (num_passed <= number) {
  471. msg_err_main ("systemd LISTEN_FDS does not contain the expected fd: %d",
  472. num_passed);
  473. errno = EINVAL;
  474. }
  475. }
  476. else {
  477. msg_err_main ("cannot get systemd variable 'LISTEN_FDS'");
  478. errno = ENOENT;
  479. }
  480. return result;
  481. }
  482. static void
  483. pass_signal_cb (gpointer key, gpointer value, gpointer ud)
  484. {
  485. struct rspamd_worker *cur = value;
  486. gint signo = GPOINTER_TO_INT (ud);
  487. kill (cur->pid, signo);
  488. }
  489. static void
  490. rspamd_pass_signal (GHashTable * workers, gint signo)
  491. {
  492. g_hash_table_foreach (workers, pass_signal_cb, GINT_TO_POINTER (signo));
  493. }
  494. static inline uintptr_t
  495. make_listen_key (struct rspamd_worker_bind_conf *cf)
  496. {
  497. rspamd_cryptobox_fast_hash_state_t st;
  498. guint i, keylen = 0;
  499. guint8 *key;
  500. rspamd_inet_addr_t *addr;
  501. guint16 port;
  502. rspamd_cryptobox_fast_hash_init (&st, rspamd_hash_seed ());
  503. if (cf->is_systemd) {
  504. /* Something like 'systemd:0' or 'systemd:controller'. */
  505. rspamd_cryptobox_fast_hash_update (&st, cf->name, strlen (cf->name));
  506. }
  507. else {
  508. rspamd_cryptobox_fast_hash_update (&st, cf->name, strlen (cf->name));
  509. for (i = 0; i < cf->cnt; i ++) {
  510. addr = g_ptr_array_index (cf->addrs, i);
  511. key = rspamd_inet_address_get_hash_key (
  512. addr, &keylen);
  513. rspamd_cryptobox_fast_hash_update (&st, key, keylen);
  514. port = rspamd_inet_address_get_port (addr);
  515. rspamd_cryptobox_fast_hash_update (&st, &port, sizeof (port));
  516. }
  517. }
  518. return rspamd_cryptobox_fast_hash_final (&st);
  519. }
  520. static void
  521. spawn_worker_type (struct rspamd_main *rspamd_main, struct ev_loop *event_loop,
  522. struct rspamd_worker_conf *cf)
  523. {
  524. gint i;
  525. if (cf->count < 0) {
  526. msg_info_main ("skip spawning of worker %s: disabled in configuration",
  527. cf->worker->name);
  528. return;
  529. }
  530. if (cf->worker->flags & RSPAMD_WORKER_UNIQUE) {
  531. if (cf->count > 1) {
  532. msg_warn_main (
  533. "cannot spawn more than 1 %s worker, so spawn one",
  534. cf->worker->name);
  535. }
  536. rspamd_fork_worker (rspamd_main, cf, 0, event_loop, rspamd_cld_handler,
  537. listen_sockets);
  538. }
  539. else if (cf->worker->flags & RSPAMD_WORKER_THREADED) {
  540. rspamd_fork_worker (rspamd_main, cf, 0, event_loop, rspamd_cld_handler,
  541. listen_sockets);
  542. }
  543. else {
  544. for (i = 0; i < cf->count; i++) {
  545. rspamd_fork_worker (rspamd_main, cf, i, event_loop,
  546. rspamd_cld_handler, listen_sockets);
  547. }
  548. }
  549. }
  550. static void
  551. spawn_workers (struct rspamd_main *rspamd_main, struct ev_loop *ev_base)
  552. {
  553. GList *cur, *ls;
  554. struct rspamd_worker_conf *cf;
  555. gpointer p;
  556. guintptr key;
  557. struct rspamd_worker_bind_conf *bcf;
  558. gboolean listen_ok = FALSE;
  559. GPtrArray *seen_mandatory_workers;
  560. worker_t **cw, *wrk;
  561. guint i;
  562. /* Special hack for hs_helper if it's not defined in a config */
  563. seen_mandatory_workers = g_ptr_array_new ();
  564. cur = rspamd_main->cfg->workers;
  565. while (cur) {
  566. cf = cur->data;
  567. listen_ok = FALSE;
  568. if (cf->worker == NULL) {
  569. msg_err_main ("type of worker is unspecified, skip spawning");
  570. }
  571. else {
  572. if (!cf->enabled || cf->count <= 0) {
  573. msg_info_main ("worker of type %s(%s) is disabled in the config, "
  574. "skip spawning", g_quark_to_string (cf->type),
  575. cf->bind_conf ? cf->bind_conf->name : "none");
  576. cur = g_list_next (cur);
  577. continue;
  578. }
  579. if (cf->worker->flags & RSPAMD_WORKER_ALWAYS_START) {
  580. g_ptr_array_add (seen_mandatory_workers, cf->worker);
  581. }
  582. if (cf->worker->flags & RSPAMD_WORKER_HAS_SOCKET) {
  583. LL_FOREACH (cf->bind_conf, bcf) {
  584. key = make_listen_key (bcf);
  585. if ((p =
  586. g_hash_table_lookup (listen_sockets,
  587. GINT_TO_POINTER (key))) == NULL) {
  588. if (!bcf->is_systemd) {
  589. /* Create listen socket */
  590. ls = create_listen_socket (bcf->addrs, bcf->cnt,
  591. cf->worker->listen_type);
  592. }
  593. else {
  594. ls = systemd_get_socket (rspamd_main,
  595. g_ptr_array_index (bcf->addrs, 0));
  596. }
  597. if (ls == NULL) {
  598. msg_err_main ("cannot listen on %s socket %s: %s",
  599. bcf->is_systemd ? "systemd" : "normal",
  600. bcf->name,
  601. strerror (errno));
  602. }
  603. else {
  604. g_hash_table_insert (listen_sockets, (gpointer)key, ls);
  605. listen_ok = TRUE;
  606. }
  607. }
  608. else {
  609. /* We had socket for this type of worker */
  610. ls = p;
  611. listen_ok = TRUE;
  612. }
  613. /* Do not add existing lists as it causes loops */
  614. if (g_list_position (cf->listen_socks, ls) == -1) {
  615. cf->listen_socks = g_list_concat (cf->listen_socks, ls);
  616. }
  617. }
  618. if (listen_ok) {
  619. spawn_worker_type (rspamd_main, ev_base, cf);
  620. }
  621. else {
  622. if (cf->bind_conf == NULL) {
  623. msg_err_main ("cannot create listen socket for %s",
  624. g_quark_to_string (cf->type));
  625. } else {
  626. msg_err_main ("cannot create listen socket for %s at %s",
  627. g_quark_to_string (cf->type), cf->bind_conf->name);
  628. }
  629. rspamd_hard_terminate (rspamd_main);
  630. g_assert_not_reached ();
  631. }
  632. }
  633. else {
  634. spawn_worker_type (rspamd_main, ev_base, cf);
  635. }
  636. }
  637. cur = g_list_next (cur);
  638. }
  639. for (cw = workers; *cw != NULL; cw ++) {
  640. gboolean seen = FALSE;
  641. wrk = *cw;
  642. if (wrk->flags & RSPAMD_WORKER_ALWAYS_START) {
  643. for (i = 0; i < seen_mandatory_workers->len; i ++) {
  644. if (wrk == g_ptr_array_index (seen_mandatory_workers, i)) {
  645. seen = TRUE;
  646. break;
  647. }
  648. }
  649. if (!seen) {
  650. cf = rspamd_config_new_worker (rspamd_main->cfg, NULL);
  651. cf->count = 1;
  652. cf->worker = wrk;
  653. cf->type = g_quark_from_static_string (wrk->name);
  654. if (cf->worker->worker_init_func) {
  655. cf->ctx = cf->worker->worker_init_func (rspamd_main->cfg);
  656. }
  657. spawn_worker_type (rspamd_main, ev_base, cf);
  658. }
  659. }
  660. }
  661. g_ptr_array_free (seen_mandatory_workers, TRUE);
  662. }
  663. static void
  664. kill_old_workers (gpointer key, gpointer value, gpointer unused)
  665. {
  666. struct rspamd_worker *w = value;
  667. struct rspamd_main *rspamd_main;
  668. rspamd_main = w->srv;
  669. if (w->state == rspamd_worker_state_wanna_die) {
  670. w->state = rspamd_worker_state_terminating;
  671. kill (w->pid, SIGUSR2);
  672. ev_io_stop (rspamd_main->event_loop, &w->srv_ev);
  673. g_hash_table_remove_all (w->control_events_pending);
  674. msg_info_main ("send signal to worker %P", w->pid);
  675. }
  676. else if (w->state != rspamd_worker_state_running) {
  677. msg_info_main ("do not send signal to worker %P, already sent", w->pid);
  678. }
  679. }
  680. static void
  681. mark_old_workers (gpointer key, gpointer value, gpointer unused)
  682. {
  683. struct rspamd_worker *w = value;
  684. struct rspamd_main *rspamd_main;
  685. rspamd_main = w->srv;
  686. if (w->state == rspamd_worker_state_running) {
  687. w->state = rspamd_worker_state_wanna_die;
  688. }
  689. }
  690. static void
  691. rspamd_worker_wait (struct rspamd_worker *w)
  692. {
  693. struct rspamd_main *rspamd_main;
  694. rspamd_main = w->srv;
  695. if (term_attempts < 0) {
  696. if (w->cf->worker->flags & RSPAMD_WORKER_KILLABLE) {
  697. if (kill (w->pid, SIGKILL) == -1) {
  698. if (errno == ESRCH) {
  699. /* We have actually killed the process */
  700. return;
  701. }
  702. }
  703. else {
  704. msg_warn_main ("terminate worker %s(%P) with SIGKILL",
  705. g_quark_to_string (w->type), w->pid);
  706. }
  707. }
  708. else {
  709. kill (w->pid, SIGKILL);
  710. if (errno == ESRCH) {
  711. /* We have actually killed the process */
  712. return;
  713. }
  714. else {
  715. msg_err_main ("data corruption warning: terminating "
  716. "special worker %s(%P) with SIGKILL",
  717. g_quark_to_string (w->type), w->pid);
  718. }
  719. }
  720. }
  721. }
  722. static void
  723. hash_worker_wait_callback (gpointer key, gpointer value, gpointer unused)
  724. {
  725. rspamd_worker_wait ((struct rspamd_worker *)value);
  726. }
  727. struct core_check_cbdata {
  728. struct rspamd_config *cfg;
  729. gsize total_count;
  730. gsize total_size;
  731. };
  732. #ifdef HAVE_NFTW
  733. static struct core_check_cbdata cores_cbdata;
  734. static gint
  735. rspamd_check_core_cb (const gchar *path, const struct stat *st,
  736. gint flag, struct FTW *ft)
  737. {
  738. if (S_ISREG (st->st_mode)) {
  739. cores_cbdata.total_count ++;
  740. /* Use physical size instead of displayed one */
  741. cores_cbdata.total_size += st->st_blocks * 512;
  742. }
  743. return 0;
  744. }
  745. #endif
  746. static void
  747. rspamd_check_core_limits (struct rspamd_main *rspamd_main)
  748. {
  749. #ifdef HAVE_NFTW
  750. struct rspamd_config *cfg = rspamd_main->cfg;
  751. cores_cbdata.cfg = cfg;
  752. cores_cbdata.total_count = 0;
  753. cores_cbdata.total_size = 0;
  754. if (cfg->cores_dir && (cfg->max_cores_count || cfg->max_cores_size)) {
  755. if (nftw (cfg->cores_dir, rspamd_check_core_cb, 1, FTW_MOUNT|FTW_PHYS)
  756. == -1) {
  757. msg_err_main ("nftw failed for path %s: %s", cfg->cores_dir,
  758. strerror (errno));
  759. }
  760. else {
  761. if (!rspamd_main->cores_throttling) {
  762. if (cfg->max_cores_size &&
  763. cores_cbdata.total_size > cfg->max_cores_size) {
  764. msg_warn_main (
  765. "enable cores throttling as size of cores in"
  766. " %s is %Hz, limit is %Hz",
  767. cfg->cores_dir,
  768. cores_cbdata.total_size,
  769. cfg->max_cores_size);
  770. rspamd_main->cores_throttling = TRUE;
  771. }
  772. if (cfg->max_cores_count &&
  773. cores_cbdata.total_count > cfg->max_cores_count) {
  774. msg_warn_main (
  775. "enable cores throttling as count of cores in"
  776. " %s is %z, limit is %z",
  777. cfg->cores_dir,
  778. cores_cbdata.total_count,
  779. cfg->max_cores_count);
  780. rspamd_main->cores_throttling = TRUE;
  781. }
  782. }
  783. else {
  784. if (cfg->max_cores_size &&
  785. cores_cbdata.total_size < cfg->max_cores_size) {
  786. msg_info_main (
  787. "disable cores throttling as size of cores in"
  788. " %s is now %Hz, limit is %Hz",
  789. cfg->cores_dir,
  790. cores_cbdata.total_size,
  791. cfg->max_cores_size);
  792. rspamd_main->cores_throttling = FALSE;
  793. }
  794. if (cfg->max_cores_count &&
  795. cores_cbdata.total_count < cfg->max_cores_count) {
  796. msg_info_main (
  797. "disable cores throttling as count of cores in"
  798. " %s is %z, limit is %z",
  799. cfg->cores_dir,
  800. cores_cbdata.total_count,
  801. cfg->max_cores_count);
  802. rspamd_main->cores_throttling = FALSE;
  803. }
  804. }
  805. }
  806. }
  807. #endif
  808. }
  809. static void
  810. reopen_log_handler (gpointer key, gpointer value, gpointer unused)
  811. {
  812. struct rspamd_worker *w = value;
  813. struct rspamd_main *rspamd_main;
  814. rspamd_main = w->srv;
  815. if (kill (w->pid, SIGUSR1) == -1) {
  816. msg_err_main ("kill failed for pid %P: %s", w->pid, strerror (errno));
  817. }
  818. }
  819. static gboolean
  820. load_rspamd_config (struct rspamd_main *rspamd_main,
  821. struct rspamd_config *cfg, gboolean init_modules,
  822. enum rspamd_post_load_options opts,
  823. gboolean reload)
  824. {
  825. cfg->compiled_modules = modules;
  826. cfg->compiled_workers = workers;
  827. if (!rspamd_config_read (cfg, cfg->cfg_name, config_logger, rspamd_main,
  828. ucl_vars, skip_template, lua_env)) {
  829. return FALSE;
  830. }
  831. /* Strictly set temp dir */
  832. if (!cfg->temp_dir) {
  833. msg_warn_main ("tempdir is not set, trying to use $TMPDIR");
  834. cfg->temp_dir =
  835. rspamd_mempool_strdup (cfg->cfg_pool, getenv ("TMPDIR"));
  836. if (!cfg->temp_dir) {
  837. msg_warn_main ("$TMPDIR is empty too, using /tmp as default");
  838. cfg->temp_dir = rspamd_mempool_strdup (cfg->cfg_pool, "/tmp");
  839. }
  840. }
  841. if (!reload) {
  842. /*
  843. * As some rules are defined in lua, we need to process them, then init
  844. * modules and merely afterwards to init modules
  845. */
  846. rspamd_lua_post_load_config (cfg);
  847. if (init_modules) {
  848. rspamd_init_filters (cfg, reload, false);
  849. }
  850. /* Do post-load actions */
  851. rspamd_config_post_load (cfg, opts);
  852. }
  853. return TRUE;
  854. }
  855. static void
  856. rspamd_detach_worker (struct rspamd_main *rspamd_main, struct rspamd_worker *wrk)
  857. {
  858. ev_io_stop (rspamd_main->event_loop, &wrk->srv_ev);
  859. ev_timer_stop (rspamd_main->event_loop, &wrk->hb.heartbeat_ev);
  860. }
  861. static void
  862. rspamd_attach_worker (struct rspamd_main *rspamd_main, struct rspamd_worker *wrk)
  863. {
  864. ev_io_start (rspamd_main->event_loop, &wrk->srv_ev);
  865. ev_timer_start (rspamd_main->event_loop, &wrk->hb.heartbeat_ev);
  866. }
  867. static void
  868. stop_srv_ev (gpointer key, gpointer value, gpointer ud)
  869. {
  870. struct rspamd_worker *cur = (struct rspamd_worker *)value;
  871. struct rspamd_main *rspamd_main = (struct rspamd_main *)ud;
  872. rspamd_detach_worker (rspamd_main, cur);
  873. }
  874. static void
  875. start_srv_ev (gpointer key, gpointer value, gpointer ud)
  876. {
  877. struct rspamd_worker *cur = (struct rspamd_worker *)value;
  878. struct rspamd_main *rspamd_main = (struct rspamd_main *)ud;
  879. rspamd_attach_worker (rspamd_main, cur);
  880. }
  881. static void
  882. rspamd_final_timer_handler (EV_P_ ev_timer *w, int revents)
  883. {
  884. struct rspamd_main *rspamd_main = (struct rspamd_main *)w->data;
  885. term_attempts--;
  886. g_hash_table_foreach (rspamd_main->workers, hash_worker_wait_callback,
  887. NULL);
  888. if (g_hash_table_size (rspamd_main->workers) == 0) {
  889. ev_break (rspamd_main->event_loop, EVBREAK_ALL);
  890. }
  891. }
  892. /* Signal handlers */
  893. static void
  894. rspamd_term_handler (struct ev_loop *loop, ev_signal *w, int revents)
  895. {
  896. struct rspamd_main *rspamd_main = (struct rspamd_main *)w->data;
  897. static ev_timer ev_finale;
  898. ev_tstamp shutdown_ts;
  899. if (!rspamd_main->wanna_die) {
  900. rspamd_main->wanna_die = TRUE;
  901. shutdown_ts = MAX (SOFT_SHUTDOWN_TIME,
  902. rspamd_main->cfg->task_timeout * 2.0);
  903. msg_info_main ("catch termination signal, waiting for %d children for %.2f seconds",
  904. (gint)g_hash_table_size (rspamd_main->workers),
  905. valgrind_mode ? shutdown_ts * 10 : shutdown_ts);
  906. /* Stop srv events to avoid false notifications */
  907. g_hash_table_foreach (rspamd_main->workers, stop_srv_ev, rspamd_main);
  908. rspamd_pass_signal (rspamd_main->workers, SIGTERM);
  909. if (control_fd != -1) {
  910. ev_io_stop (rspamd_main->event_loop, &control_ev);
  911. close (control_fd);
  912. }
  913. if (valgrind_mode) {
  914. /* Special case if we are likely running with valgrind */
  915. term_attempts = shutdown_ts / TERMINATION_INTERVAL * 10;
  916. }
  917. else {
  918. term_attempts = shutdown_ts / TERMINATION_INTERVAL;
  919. }
  920. ev_finale.data = rspamd_main;
  921. ev_timer_init (&ev_finale, rspamd_final_timer_handler,
  922. TERMINATION_INTERVAL, TERMINATION_INTERVAL);
  923. ev_timer_start (rspamd_main->event_loop, &ev_finale);
  924. }
  925. }
  926. static void
  927. rspamd_usr1_handler (struct ev_loop *loop, ev_signal *w, int revents)
  928. {
  929. struct rspamd_main *rspamd_main = (struct rspamd_main *)w->data;
  930. if (!rspamd_main->wanna_die) {
  931. rspamd_log_reopen (rspamd_main->logger,
  932. rspamd_main->cfg,
  933. rspamd_main->workers_uid,
  934. rspamd_main->workers_gid);
  935. msg_info_main ("logging reinitialised");
  936. g_hash_table_foreach (rspamd_main->workers, reopen_log_handler,
  937. NULL);
  938. }
  939. }
  940. static void
  941. rspamd_stat_update_handler (struct ev_loop *loop, ev_timer *w, int revents)
  942. {
  943. struct rspamd_main *rspamd_main = (struct rspamd_main *)w->data;
  944. struct rspamd_stat cur_stat;
  945. gchar proctitle[128];
  946. memcpy (&cur_stat, rspamd_main->stat, sizeof (cur_stat));
  947. if (old_stat.messages_scanned > 0 &&
  948. cur_stat.messages_scanned > old_stat.messages_scanned) {
  949. gdouble rate = (double)(cur_stat.messages_scanned - old_stat.messages_scanned) /
  950. w->repeat;
  951. gdouble old_spam = old_stat.actions_stat[METRIC_ACTION_REJECT] +
  952. old_stat.actions_stat[METRIC_ACTION_ADD_HEADER] +
  953. old_stat.actions_stat[METRIC_ACTION_REWRITE_SUBJECT];
  954. gdouble old_ham = old_stat.actions_stat[METRIC_ACTION_NOACTION];
  955. gdouble new_spam = cur_stat.actions_stat[METRIC_ACTION_REJECT] +
  956. cur_stat.actions_stat[METRIC_ACTION_ADD_HEADER] +
  957. cur_stat.actions_stat[METRIC_ACTION_REWRITE_SUBJECT];
  958. gdouble new_ham = cur_stat.actions_stat[METRIC_ACTION_NOACTION];
  959. rspamd_snprintf (proctitle, sizeof (proctitle),
  960. "main process; %.1f msg/sec, %.1f msg/sec spam, %.1f msg/sec ham",
  961. rate,
  962. (new_spam - old_spam) / w->repeat,
  963. (new_ham - old_ham) / w->repeat);
  964. setproctitle (proctitle);
  965. }
  966. memcpy (&old_stat, &cur_stat, sizeof (cur_stat));
  967. }
  968. static void
  969. rspamd_hup_handler (struct ev_loop *loop, ev_signal *w, int revents)
  970. {
  971. struct rspamd_main *rspamd_main = (struct rspamd_main *)w->data;
  972. if (!rspamd_main->wanna_die) {
  973. msg_info_main ("rspamd "
  974. RVERSION
  975. " is requested to reload configuration");
  976. /* Detach existing workers and stop their heartbeats */
  977. g_hash_table_foreach (rspamd_main->workers, stop_srv_ev, rspamd_main);
  978. if (reread_config (rspamd_main)) {
  979. rspamd_check_core_limits (rspamd_main);
  980. /* Mark old workers */
  981. g_hash_table_foreach (rspamd_main->workers, mark_old_workers, NULL);
  982. msg_info_main ("spawn workers with a new config");
  983. spawn_workers (rspamd_main, rspamd_main->event_loop);
  984. msg_info_main ("workers spawning has been finished");
  985. /* Kill marked */
  986. msg_info_main ("kill old workers");
  987. g_hash_table_foreach (rspamd_main->workers, kill_old_workers, NULL);
  988. }
  989. else {
  990. /* Reattach old workers */
  991. msg_info_main ("restore old workers with a old config");
  992. g_hash_table_foreach (rspamd_main->workers, start_srv_ev, rspamd_main);
  993. }
  994. }
  995. }
  996. /* Called when a dead child has been found */
  997. static void
  998. rspamd_cld_handler (EV_P_ ev_child *w, struct rspamd_main *rspamd_main,
  999. struct rspamd_worker *wrk)
  1000. {
  1001. gboolean need_refork;
  1002. static struct rspamd_control_command cmd;
  1003. /* Turn off locking for logger */
  1004. ev_child_stop (EV_A_ w);
  1005. /* Remove dead child form children list */
  1006. g_hash_table_remove (rspamd_main->workers, GSIZE_TO_POINTER (wrk->pid));
  1007. g_hash_table_remove_all (wrk->control_events_pending);
  1008. if (wrk->srv_pipe[0] != -1) {
  1009. /* Ugly workaround */
  1010. if (wrk->tmp_data) {
  1011. g_free (wrk->tmp_data);
  1012. }
  1013. rspamd_detach_worker (rspamd_main, wrk);
  1014. }
  1015. if (wrk->control_pipe[0] != -1) {
  1016. /* We also need to clean descriptors left */
  1017. close (wrk->control_pipe[0]);
  1018. close (wrk->srv_pipe[0]);
  1019. }
  1020. if (!rspamd_main->wanna_die) {
  1021. cmd.type = RSPAMD_CONTROL_CHILD_CHANGE;
  1022. cmd.cmd.child_change.what = rspamd_child_terminated;
  1023. cmd.cmd.child_change.pid = wrk->pid;
  1024. cmd.cmd.child_change.additional = w->rstatus;
  1025. rspamd_control_broadcast_srv_cmd (rspamd_main, &cmd, wrk->pid);
  1026. }
  1027. need_refork = rspamd_check_termination_clause (wrk->srv, wrk, w->rstatus);
  1028. if (need_refork) {
  1029. /* Fork another worker in replace of dead one */
  1030. msg_info_main ("respawn process %s in lieu of terminated process with pid %P",
  1031. g_quark_to_string (wrk->type),
  1032. wrk->pid);
  1033. rspamd_check_core_limits (rspamd_main);
  1034. rspamd_fork_delayed (wrk->cf, wrk->index, rspamd_main);
  1035. }
  1036. else {
  1037. msg_info_main ("do not respawn process %s after found terminated process with pid %P",
  1038. g_quark_to_string (wrk->type),
  1039. wrk->pid);
  1040. }
  1041. REF_RELEASE (wrk->cf);
  1042. g_hash_table_unref (wrk->control_events_pending);
  1043. g_free (wrk);
  1044. }
  1045. /* Control socket handler */
  1046. static void
  1047. rspamd_control_handler (EV_P_ ev_io *w, int revents)
  1048. {
  1049. struct rspamd_main *rspamd_main = (struct rspamd_main *)w->data;
  1050. rspamd_inet_addr_t *addr;
  1051. gint nfd;
  1052. if ((nfd =
  1053. rspamd_accept_from_socket (w->fd, &addr, NULL, NULL)) == -1) {
  1054. msg_warn_main ("accept failed: %s", strerror (errno));
  1055. return;
  1056. }
  1057. /* Check for EAGAIN */
  1058. if (nfd == 0) {
  1059. return;
  1060. }
  1061. msg_info_main ("accepted control connection from %s",
  1062. rspamd_inet_address_to_string (addr));
  1063. rspamd_control_process_client_socket (rspamd_main, nfd, addr);
  1064. }
  1065. static guint
  1066. rspamd_spair_hash (gconstpointer p)
  1067. {
  1068. return rspamd_cryptobox_fast_hash (p, PAIR_ID_LEN, rspamd_hash_seed ());
  1069. }
  1070. static gboolean
  1071. rspamd_spair_equal (gconstpointer a, gconstpointer b)
  1072. {
  1073. return memcmp (a, b, PAIR_ID_LEN) == 0;
  1074. }
  1075. static void
  1076. rspamd_spair_close (gpointer p)
  1077. {
  1078. gint *fds = p;
  1079. close (fds[0]);
  1080. close (fds[1]);
  1081. g_free (p);
  1082. }
  1083. static void
  1084. version (void)
  1085. {
  1086. #if defined(GIT_VERSION) && GIT_VERSION == 1
  1087. rspamd_printf ("Rspamd daemon version " RVERSION "-git." RID "\n");
  1088. #else
  1089. rspamd_printf ("Rspamd daemon version " RVERSION "\n");
  1090. #endif
  1091. }
  1092. gint
  1093. main (gint argc, gchar **argv, gchar **env)
  1094. {
  1095. gint i, res = 0;
  1096. struct sigaction signals, sigpipe_act;
  1097. worker_t **pworker;
  1098. GQuark type;
  1099. rspamd_inet_addr_t *control_addr = NULL;
  1100. struct ev_loop *event_loop;
  1101. struct rspamd_main *rspamd_main;
  1102. gboolean skip_pid = FALSE;
  1103. rspamd_main = (struct rspamd_main *) g_malloc0 (sizeof (struct rspamd_main));
  1104. rspamd_main->server_pool = rspamd_mempool_new (rspamd_mempool_suggest_size (),
  1105. "main", 0);
  1106. rspamd_main->stat = rspamd_mempool_alloc0_shared (rspamd_main->server_pool,
  1107. sizeof (struct rspamd_stat));
  1108. rspamd_main->cfg = rspamd_config_new (RSPAMD_CONFIG_INIT_DEFAULT);
  1109. rspamd_main->spairs = g_hash_table_new_full (rspamd_spair_hash,
  1110. rspamd_spair_equal, g_free, rspamd_spair_close);
  1111. rspamd_main->start_mtx = rspamd_mempool_get_mutex (rspamd_main->server_pool);
  1112. if (getenv ("VALGRIND") != NULL) {
  1113. valgrind_mode = TRUE;
  1114. }
  1115. #ifndef HAVE_SETPROCTITLE
  1116. init_title (rspamd_main->server_pool, argc, argv, env);
  1117. #endif
  1118. rspamd_main->cfg->libs_ctx = rspamd_init_libs ();
  1119. memset (&signals, 0, sizeof (struct sigaction));
  1120. read_cmd_line (&argc, &argv, rspamd_main->cfg);
  1121. if (show_version) {
  1122. version ();
  1123. exit (EXIT_SUCCESS);
  1124. }
  1125. if (argc > 0) {
  1126. /* Parse variables */
  1127. for (i = 0; i < argc; i++) {
  1128. if (strchr (argv[i], '=') != NULL) {
  1129. gchar *k, *v, *t;
  1130. k = g_strdup (argv[i]);
  1131. t = strchr (k, '=');
  1132. v = g_strdup (t + 1);
  1133. *t = '\0';
  1134. if (ucl_vars == NULL) {
  1135. ucl_vars = g_hash_table_new_full (rspamd_strcase_hash,
  1136. rspamd_strcase_equal, g_free, g_free);
  1137. }
  1138. g_hash_table_insert (ucl_vars, k, v);
  1139. }
  1140. }
  1141. }
  1142. if (is_debug) {
  1143. rspamd_main->cfg->log_level = G_LOG_LEVEL_DEBUG;
  1144. }
  1145. else {
  1146. rspamd_main->cfg->log_level = G_LOG_LEVEL_MESSAGE;
  1147. }
  1148. type = g_quark_from_static_string ("main");
  1149. /* First set logger to console logger */
  1150. rspamd_main->logger = rspamd_log_open_emergency (rspamd_main->server_pool);
  1151. g_assert (rspamd_main->logger != NULL);
  1152. if (is_debug) {
  1153. rspamd_log_set_log_level (rspamd_main->logger, G_LOG_LEVEL_DEBUG);
  1154. }
  1155. else {
  1156. rspamd_log_set_log_level (rspamd_main->logger, G_LOG_LEVEL_MESSAGE);
  1157. }
  1158. g_log_set_default_handler (rspamd_glib_log_function, rspamd_main->logger);
  1159. g_set_printerr_handler (rspamd_glib_printerr_function);
  1160. detect_priv (rspamd_main);
  1161. pworker = &workers[0];
  1162. while (*pworker) {
  1163. /* Init string quarks */
  1164. (void) g_quark_from_static_string ((*pworker)->name);
  1165. pworker++;
  1166. }
  1167. /* Init listen sockets hash */
  1168. listen_sockets = g_hash_table_new (g_direct_hash, g_direct_equal);
  1169. sqlite3_initialize ();
  1170. /* Load config */
  1171. if (!load_rspamd_config (rspamd_main, rspamd_main->cfg, TRUE,
  1172. RSPAMD_CONFIG_LOAD_ALL, FALSE)) {
  1173. exit (EXIT_FAILURE);
  1174. }
  1175. /* Override pidfile from configuration by command line argument */
  1176. if (rspamd_pidfile != NULL) {
  1177. rspamd_main->cfg->pid_file = rspamd_pidfile;
  1178. }
  1179. /* Force debug log */
  1180. if (is_debug) {
  1181. rspamd_log_set_log_level (rspamd_main->logger, G_LOG_LEVEL_DEBUG);
  1182. }
  1183. /* Create rolling history */
  1184. rspamd_main->history = rspamd_roll_history_new (rspamd_main->server_pool,
  1185. rspamd_main->cfg->history_rows, rspamd_main->cfg);
  1186. msg_info_main ("rspamd "
  1187. RVERSION
  1188. " is starting, build id: "
  1189. RID);
  1190. rspamd_main->cfg->cfg_name = rspamd_mempool_strdup (
  1191. rspamd_main->cfg->cfg_pool,
  1192. rspamd_main->cfg->cfg_name);
  1193. msg_info_main ("cpu features: %s",
  1194. rspamd_main->cfg->libs_ctx->crypto_ctx->cpu_extensions);
  1195. msg_info_main ("cryptobox configuration: curve25519(libsodium), "
  1196. "chacha20(%s), poly1305(libsodium), siphash(libsodium), blake2(libsodium), base64(%s)",
  1197. rspamd_main->cfg->libs_ctx->crypto_ctx->chacha20_impl,
  1198. rspamd_main->cfg->libs_ctx->crypto_ctx->base64_impl);
  1199. msg_info_main ("libottery prf: %s", ottery_get_impl_name ());
  1200. /* Daemonize */
  1201. if (!no_fork) {
  1202. if (daemon (0, 0) == -1) {
  1203. msg_err_main ("cannot daemonize: %s", strerror (errno));
  1204. exit (-errno);
  1205. }
  1206. /* Close emergency logger */
  1207. rspamd_log_close (rspamd_log_emergency_logger ());
  1208. }
  1209. /* Write info */
  1210. rspamd_main->pid = getpid ();
  1211. rspamd_main->type = type;
  1212. if (!valgrind_mode) {
  1213. rspamd_set_crash_handler (rspamd_main);
  1214. }
  1215. /* Ignore SIGPIPE as we handle write errors manually */
  1216. sigemptyset (&sigpipe_act.sa_mask);
  1217. sigaddset (&sigpipe_act.sa_mask, SIGPIPE);
  1218. sigpipe_act.sa_handler = SIG_IGN;
  1219. sigpipe_act.sa_flags = 0;
  1220. sigaction (SIGPIPE, &sigpipe_act, NULL);
  1221. if (rspamd_main->cfg->pid_file == NULL) {
  1222. msg_info_main ("pid file is not specified, skipping writing it");
  1223. skip_pid = TRUE;
  1224. }
  1225. else if (no_fork) {
  1226. msg_info_main ("skip writing pid in no-fork mode");
  1227. skip_pid = TRUE;
  1228. }
  1229. else if (rspamd_write_pid (rspamd_main) == -1) {
  1230. msg_err_main ("cannot write pid file %s", rspamd_main->cfg->pid_file);
  1231. exit (-errno);
  1232. }
  1233. /* Block signals to use sigsuspend in future */
  1234. sigprocmask (SIG_BLOCK, &signals.sa_mask, NULL);
  1235. /* Set title */
  1236. setproctitle ("main process");
  1237. /* Open control socket if needed */
  1238. control_fd = -1;
  1239. if (rspamd_main->cfg->control_socket_path) {
  1240. if (!rspamd_parse_inet_address (&control_addr,
  1241. rspamd_main->cfg->control_socket_path,
  1242. strlen (rspamd_main->cfg->control_socket_path),
  1243. RSPAMD_INET_ADDRESS_PARSE_DEFAULT)) {
  1244. msg_err_main ("cannot parse inet address %s",
  1245. rspamd_main->cfg->control_socket_path);
  1246. }
  1247. else {
  1248. control_fd = rspamd_inet_address_listen (control_addr, SOCK_STREAM,
  1249. RSPAMD_INET_ADDRESS_LISTEN_ASYNC, -1);
  1250. if (control_fd == -1) {
  1251. msg_err_main ("cannot open control socket at path: %s",
  1252. rspamd_main->cfg->control_socket_path);
  1253. }
  1254. }
  1255. }
  1256. /* Maybe read roll history */
  1257. if (rspamd_main->cfg->history_file) {
  1258. rspamd_roll_history_load (rspamd_main->history,
  1259. rspamd_main->cfg->history_file);
  1260. }
  1261. /* Spawn workers */
  1262. rspamd_main->workers = g_hash_table_new (g_direct_hash, g_direct_equal);
  1263. /* Init event base */
  1264. event_loop = ev_default_loop (rspamd_config_ev_backend_get (rspamd_main->cfg));
  1265. rspamd_main->event_loop = event_loop;
  1266. if (event_loop) {
  1267. int loop_type = ev_backend (event_loop);
  1268. gboolean effective_backend;
  1269. const gchar *loop_str;
  1270. loop_str =
  1271. rspamd_config_ev_backend_to_string (loop_type, &effective_backend);
  1272. if (!effective_backend) {
  1273. msg_warn_main ("event loop uses non-optimal backend: %s", loop_str);
  1274. }
  1275. else {
  1276. msg_info_main ("event loop initialised with backend: %s", loop_str);
  1277. }
  1278. }
  1279. else {
  1280. msg_err ("cannot init event loop! exiting");
  1281. exit (EXIT_FAILURE);
  1282. }
  1283. /* Unblock signals */
  1284. sigemptyset (&signals.sa_mask);
  1285. sigprocmask (SIG_SETMASK, &signals.sa_mask, NULL);
  1286. /* Set events for signals */
  1287. ev_signal_init (&rspamd_main->term_ev, rspamd_term_handler, SIGTERM);
  1288. rspamd_main->term_ev.data = rspamd_main;
  1289. ev_signal_start (event_loop, &rspamd_main->term_ev);
  1290. ev_signal_init (&rspamd_main->int_ev, rspamd_term_handler, SIGINT);
  1291. rspamd_main->int_ev.data = rspamd_main;
  1292. ev_signal_start (event_loop, &rspamd_main->int_ev);
  1293. ev_signal_init (&rspamd_main->hup_ev, rspamd_hup_handler, SIGHUP);
  1294. rspamd_main->hup_ev.data = rspamd_main;
  1295. ev_signal_start (event_loop, &rspamd_main->hup_ev);
  1296. ev_signal_init (&rspamd_main->usr1_ev, rspamd_usr1_handler, SIGUSR1);
  1297. rspamd_main->usr1_ev.data = rspamd_main;
  1298. ev_signal_start (event_loop, &rspamd_main->usr1_ev);
  1299. /* Update proctitle according to number of messages processed */
  1300. static const ev_tstamp stat_update_time = 10.0;
  1301. memset (&old_stat, 0, sizeof (old_stat));
  1302. stat_ev.data = rspamd_main;
  1303. ev_timer_init (&stat_ev, rspamd_stat_update_handler,
  1304. stat_update_time, stat_update_time);
  1305. ev_timer_start (event_loop, &stat_ev);
  1306. rspamd_check_core_limits (rspamd_main);
  1307. rspamd_mempool_lock_mutex (rspamd_main->start_mtx);
  1308. spawn_workers (rspamd_main, event_loop);
  1309. rspamd_mempool_unlock_mutex (rspamd_main->start_mtx);
  1310. rspamd_main->http_ctx = rspamd_http_context_create (rspamd_main->cfg,
  1311. event_loop, rspamd_main->cfg->ups_ctx);
  1312. if (control_fd != -1) {
  1313. msg_info_main ("listening for control commands on %s",
  1314. rspamd_inet_address_to_string (control_addr));
  1315. ev_io_init (&control_ev, rspamd_control_handler, control_fd, EV_READ);
  1316. control_ev.data = rspamd_main;
  1317. ev_io_start (event_loop, &control_ev);
  1318. }
  1319. ev_loop (event_loop, 0);
  1320. /* Maybe save roll history */
  1321. if (rspamd_main->cfg->history_file) {
  1322. rspamd_roll_history_save (rspamd_main->history,
  1323. rspamd_main->cfg->history_file);
  1324. }
  1325. msg_info_main ("terminating...");
  1326. REF_RELEASE (rspamd_main->cfg);
  1327. rspamd_log_close (rspamd_main->logger);
  1328. g_hash_table_unref (rspamd_main->spairs);
  1329. g_hash_table_unref (rspamd_main->workers);
  1330. rspamd_mempool_delete (rspamd_main->server_pool);
  1331. if (!skip_pid) {
  1332. rspamd_pidfile_close (rspamd_main->pfh);
  1333. }
  1334. g_free (rspamd_main);
  1335. ev_unref (event_loop);
  1336. sqlite3_shutdown ();
  1337. if (control_addr) {
  1338. rspamd_inet_address_free (control_addr);
  1339. }
  1340. return (res);
  1341. }