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

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697989910010110210310410510610710810911011111211311411511611711811912012112212312412512612712812913013113213313413513613713813914014114214314414514614714814915015115215315415515615715815916016116216316416516616716816917017117217317417517617717817918018118218318418518618718818919019119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723823924024124224324424524624724824925025125225325425525625725825926026126226326426526626726826927027127227327427527627727827928028128228328428528628728828929029129229329429529629729829930030130230330430530630730830931031131231331431531631731831932032132232332432532632732832933033133233333433533633733833934034134234334434534634734834935035135235335435535635735835936036136236336436536636736836937037137237337437537637737837938038138238338438538638738838939039139239339439539639739839940040140240340440540640740840941041141241341441541641741841942042142242342442542642742842943043143243343443543643743843944044144244344444544644744844945045145245345445545645745845946046146246346446546646746846947047147247347447547647747847948048148248348448548648748848949049149249349449549649749849950050150250350450550650750850951051151251351451551651751851952052152252352452552652752852953053153253353453553653753853954054154254354454554654754854955055155255355455555655755855956056156256356456556656756856957057157257357457557657757857958058158258358458558658758858959059159259359459559659759859960060160260360460560660760860961061161261361461561661761861962062162262362462562662762862963063163263363463563663763863964064164264364464564664764864965065165265365465565665765865966066166266366466566666766866967067167267367467567667767867968068168268368468568668768868969069169269369469569669769869970070170270370470570670770870971071171271371471571671771871972072172272372472572672772872973073173273373473573673773873974074174274374474574674774874975075175275375475575675775875976076176276376476576676776876977077177277377477577677777877978078178278378478578678778878979079179279379479579679779879980080180280380480580680780880981081181281381481581681781881982082182282382482582682782882983083183283383483583683783883984084184284384484584684784884985085185285385485585685785885986086186286386486586686786886987087187287387487587687787887988088188288388488588688788888989089189289389489589689789889990090190290390490590690790890991091191291391491591691791891992092192292392492592692792892993093193293393493593693793893994094194294394494594694794894995095195295395495595695795895996096196296396496596696796896997097197297397497597697797897998098198298398498598698798898999099199299399499599699799899910001001100210031004100510061007100810091010101110121013101410151016101710181019102010211022102310241025102610271028102910301031103210331034103510361037103810391040104110421043104410451046104710481049105010511052105310541055105610571058105910601061106210631064106510661067106810691070107110721073107410751076107710781079108010811082108310841085108610871088108910901091109210931094109510961097109810991100110111021103110411051106110711081109111011111112111311141115111611171118111911201121112211231124112511261127112811291130113111321133113411351136113711381139114011411142114311441145114611471148114911501151115211531154115511561157115811591160116111621163116411651166116711681169117011711172117311741175117611771178117911801181118211831184118511861187118811891190119111921193119411951196119711981199120012011202120312041205120612071208120912101211121212131214121512161217121812191220122112221223122412251226122712281229123012311232123312341235123612371238123912401241124212431244124512461247124812491250125112521253125412551256125712581259126012611262126312641265126612671268126912701271127212731274127512761277127812791280128112821283128412851286128712881289129012911292129312941295129612971298129913001301130213031304130513061307130813091310131113121313131413151316131713181319132013211322132313241325132613271328132913301331133213331334133513361337133813391340134113421343134413451346134713481349135013511352135313541355135613571358135913601361136213631364136513661367136813691370137113721373137413751376137713781379138013811382138313841385138613871388138913901391139213931394139513961397139813991400140114021403140414051406140714081409141014111412141314141415141614171418141914201421142214231424142514261427142814291430143114321433143414351436143714381439144014411442144314441445144614471448144914501451145214531454145514561457145814591460146114621463146414651466146714681469147014711472147314741475147614771478147914801481148214831484148514861487148814891490149114921493149414951496149714981499150015011502150315041505150615071508150915101511151215131514151515161517151815191520152115221523152415251526152715281529153015311532153315341535153615371538153915401541154215431544154515461547154815491550155115521553155415551556
  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. RSPAMD_INET_ADDRESS_LISTEN_REUSEPORT|
  350. RSPAMD_INET_ADDRESS_LISTEN_NOLISTEN;
  351. struct rspamd_worker_listen_socket *ls;
  352. g_ptr_array_sort (addrs, rspamd_inet_address_compare_ptr);
  353. for (i = 0; i < cnt; i ++) {
  354. /*
  355. * Copy address to avoid reload issues
  356. */
  357. if (listen_type & RSPAMD_WORKER_SOCKET_TCP) {
  358. fd = rspamd_inet_address_listen (g_ptr_array_index (addrs, i),
  359. SOCK_STREAM,
  360. listen_opts, -1);
  361. if (fd != -1) {
  362. ls = g_malloc0 (sizeof (*ls));
  363. ls->addr = rspamd_inet_address_copy (g_ptr_array_index (addrs, i));
  364. ls->fd = fd;
  365. ls->type = RSPAMD_WORKER_SOCKET_TCP;
  366. result = g_list_prepend (result, ls);
  367. }
  368. }
  369. if (listen_type & RSPAMD_WORKER_SOCKET_UDP) {
  370. fd = rspamd_inet_address_listen (g_ptr_array_index (addrs, i),
  371. SOCK_DGRAM,
  372. listen_opts, -1);
  373. if (fd != -1) {
  374. ls = g_malloc0 (sizeof (*ls));
  375. ls->addr = rspamd_inet_address_copy (g_ptr_array_index (addrs, i));
  376. ls->fd = fd;
  377. ls->type = RSPAMD_WORKER_SOCKET_UDP;
  378. result = g_list_prepend (result, ls);
  379. }
  380. }
  381. }
  382. return result;
  383. }
  384. static GList *
  385. systemd_get_socket (struct rspamd_main *rspamd_main, const gchar *fdname)
  386. {
  387. int number, sock, num_passed, flags;
  388. GList *result = NULL;
  389. const gchar *e;
  390. gchar **fdnames;
  391. gchar *end;
  392. struct stat st;
  393. static const int sd_listen_fds_start = 3; /* SD_LISTEN_FDS_START */
  394. struct rspamd_worker_listen_socket *ls;
  395. union {
  396. struct sockaddr_storage ss;
  397. struct sockaddr sa;
  398. } addr_storage;
  399. socklen_t slen = sizeof (addr_storage);
  400. gint stype;
  401. number = strtoul (fdname, &end, 10);
  402. if (end != NULL && *end != '\0') {
  403. /* Cannot parse as number, assume a name in LISTEN_FDNAMES. */
  404. e = getenv ("LISTEN_FDNAMES");
  405. if (!e) {
  406. msg_err_main ("cannot get systemd variable 'LISTEN_FDNAMES'");
  407. errno = ENOENT;
  408. return NULL;
  409. }
  410. fdnames = g_strsplit (e, ":", -1);
  411. for (number = 0; fdnames[number]; number++) {
  412. if (!strcmp (fdnames[number], fdname)) {
  413. break;
  414. }
  415. }
  416. if (!fdnames[number]) {
  417. number = -1;
  418. }
  419. g_strfreev (fdnames);
  420. }
  421. if (number < 0) {
  422. msg_warn_main ("cannot find systemd socket: %s", fdname);
  423. errno = ENOENT;
  424. return NULL;
  425. }
  426. e = getenv ("LISTEN_FDS");
  427. if (e != NULL) {
  428. errno = 0;
  429. num_passed = strtoul (e, &end, 10);
  430. if ((end == NULL || *end == '\0') && num_passed > number) {
  431. sock = number + sd_listen_fds_start;
  432. if (fstat (sock, &st) == -1) {
  433. msg_warn_main ("cannot stat systemd descriptor %d", sock);
  434. return NULL;
  435. }
  436. if (!S_ISSOCK (st.st_mode)) {
  437. msg_warn_main ("systemd descriptor %d is not a socket", sock);
  438. errno = EINVAL;
  439. return NULL;
  440. }
  441. flags = fcntl (sock, F_GETFD);
  442. if (flags != -1) {
  443. (void)fcntl (sock, F_SETFD, flags | FD_CLOEXEC);
  444. }
  445. rspamd_socket_nonblocking (sock);
  446. if (getsockname (sock, &addr_storage.sa, &slen) == -1) {
  447. msg_warn_main ("cannot get name for systemd descriptor %d: %s",
  448. sock, strerror (errno));
  449. errno = EINVAL;
  450. return NULL;
  451. }
  452. ls = g_malloc0 (sizeof (*ls));
  453. ls->addr = rspamd_inet_address_from_sa (&addr_storage.sa, slen);
  454. ls->fd = sock;
  455. ls->is_systemd = true;
  456. slen = sizeof (stype);
  457. if (getsockopt (sock, SOL_SOCKET, SO_TYPE, &stype, &slen) != -1) {
  458. if (stype == SOCK_STREAM) {
  459. ls->type = RSPAMD_WORKER_SOCKET_TCP;
  460. }
  461. else {
  462. ls->type = RSPAMD_WORKER_SOCKET_UDP;
  463. }
  464. }
  465. else {
  466. msg_warn_main ("cannot get type for systemd descriptor %d: %s",
  467. sock, strerror (errno));
  468. ls->type = RSPAMD_WORKER_SOCKET_TCP;
  469. }
  470. result = g_list_prepend (result, ls);
  471. }
  472. else if (num_passed <= number) {
  473. msg_err_main ("systemd LISTEN_FDS does not contain the expected fd: %d",
  474. num_passed);
  475. errno = EINVAL;
  476. }
  477. }
  478. else {
  479. msg_err_main ("cannot get systemd variable 'LISTEN_FDS'");
  480. errno = ENOENT;
  481. }
  482. return result;
  483. }
  484. static void
  485. pass_signal_cb (gpointer key, gpointer value, gpointer ud)
  486. {
  487. struct rspamd_worker *cur = value;
  488. gint signo = GPOINTER_TO_INT (ud);
  489. kill (cur->pid, signo);
  490. }
  491. static void
  492. rspamd_pass_signal (GHashTable * workers, gint signo)
  493. {
  494. g_hash_table_foreach (workers, pass_signal_cb, GINT_TO_POINTER (signo));
  495. }
  496. static inline uintptr_t
  497. make_listen_key (struct rspamd_worker_bind_conf *cf)
  498. {
  499. rspamd_cryptobox_fast_hash_state_t st;
  500. guint i, keylen = 0;
  501. guint8 *key;
  502. rspamd_inet_addr_t *addr;
  503. guint16 port;
  504. rspamd_cryptobox_fast_hash_init (&st, rspamd_hash_seed ());
  505. if (cf->is_systemd) {
  506. /* Something like 'systemd:0' or 'systemd:controller'. */
  507. rspamd_cryptobox_fast_hash_update (&st, cf->name, strlen (cf->name));
  508. }
  509. else {
  510. rspamd_cryptobox_fast_hash_update (&st, cf->name, strlen (cf->name));
  511. for (i = 0; i < cf->cnt; i ++) {
  512. addr = g_ptr_array_index (cf->addrs, i);
  513. key = rspamd_inet_address_get_hash_key (
  514. addr, &keylen);
  515. rspamd_cryptobox_fast_hash_update (&st, key, keylen);
  516. port = rspamd_inet_address_get_port (addr);
  517. rspamd_cryptobox_fast_hash_update (&st, &port, sizeof (port));
  518. }
  519. }
  520. return rspamd_cryptobox_fast_hash_final (&st);
  521. }
  522. static void
  523. spawn_worker_type (struct rspamd_main *rspamd_main, struct ev_loop *event_loop,
  524. struct rspamd_worker_conf *cf)
  525. {
  526. gint i;
  527. if (cf->count < 0) {
  528. msg_info_main ("skip spawning of worker %s: disabled in configuration",
  529. cf->worker->name);
  530. return;
  531. }
  532. if (cf->worker->flags & RSPAMD_WORKER_UNIQUE) {
  533. if (cf->count > 1) {
  534. msg_warn_main (
  535. "cannot spawn more than 1 %s worker, so spawn one",
  536. cf->worker->name);
  537. }
  538. rspamd_fork_worker (rspamd_main, cf, 0, event_loop, rspamd_cld_handler,
  539. listen_sockets);
  540. }
  541. else if (cf->worker->flags & RSPAMD_WORKER_THREADED) {
  542. rspamd_fork_worker (rspamd_main, cf, 0, event_loop, rspamd_cld_handler,
  543. listen_sockets);
  544. }
  545. else {
  546. for (i = 0; i < cf->count; i++) {
  547. rspamd_fork_worker (rspamd_main, cf, i, event_loop,
  548. rspamd_cld_handler, listen_sockets);
  549. }
  550. }
  551. }
  552. static void
  553. spawn_workers (struct rspamd_main *rspamd_main, struct ev_loop *ev_base)
  554. {
  555. GList *cur, *ls;
  556. struct rspamd_worker_conf *cf;
  557. gpointer p;
  558. guintptr key;
  559. struct rspamd_worker_bind_conf *bcf;
  560. gboolean listen_ok = FALSE;
  561. GPtrArray *seen_mandatory_workers;
  562. worker_t **cw, *wrk;
  563. guint i;
  564. /* Special hack for hs_helper if it's not defined in a config */
  565. seen_mandatory_workers = g_ptr_array_new ();
  566. cur = rspamd_main->cfg->workers;
  567. while (cur) {
  568. cf = cur->data;
  569. listen_ok = FALSE;
  570. if (cf->worker == NULL) {
  571. msg_err_main ("type of worker is unspecified, skip spawning");
  572. }
  573. else {
  574. if (!cf->enabled || cf->count <= 0) {
  575. msg_info_main ("worker of type %s(%s) is disabled in the config, "
  576. "skip spawning", g_quark_to_string (cf->type),
  577. cf->bind_conf ? cf->bind_conf->name : "none");
  578. cur = g_list_next (cur);
  579. continue;
  580. }
  581. if (cf->worker->flags & RSPAMD_WORKER_ALWAYS_START) {
  582. g_ptr_array_add (seen_mandatory_workers, cf->worker);
  583. }
  584. if (cf->worker->flags & RSPAMD_WORKER_HAS_SOCKET) {
  585. LL_FOREACH (cf->bind_conf, bcf) {
  586. key = make_listen_key (bcf);
  587. if ((p =
  588. g_hash_table_lookup (listen_sockets,
  589. GINT_TO_POINTER (key))) == NULL) {
  590. if (!bcf->is_systemd) {
  591. /* Create listen socket */
  592. ls = create_listen_socket (bcf->addrs, bcf->cnt,
  593. cf->worker->listen_type);
  594. }
  595. else {
  596. ls = systemd_get_socket (rspamd_main,
  597. g_ptr_array_index (bcf->addrs, 0));
  598. }
  599. if (ls == NULL) {
  600. msg_err_main ("cannot listen on %s socket %s: %s",
  601. bcf->is_systemd ? "systemd" : "normal",
  602. bcf->name,
  603. strerror (errno));
  604. }
  605. else {
  606. g_hash_table_insert (listen_sockets, (gpointer)key, ls);
  607. listen_ok = TRUE;
  608. }
  609. }
  610. else {
  611. /* We had socket for this type of worker */
  612. ls = p;
  613. listen_ok = TRUE;
  614. }
  615. /* Do not add existing lists as it causes loops */
  616. if (g_list_position (cf->listen_socks, ls) == -1) {
  617. cf->listen_socks = g_list_concat (cf->listen_socks, ls);
  618. }
  619. }
  620. if (listen_ok) {
  621. spawn_worker_type (rspamd_main, ev_base, cf);
  622. }
  623. else {
  624. if (cf->bind_conf == NULL) {
  625. msg_err_main ("cannot create listen socket for %s",
  626. g_quark_to_string (cf->type));
  627. } else {
  628. msg_err_main ("cannot create listen socket for %s at %s",
  629. g_quark_to_string (cf->type), cf->bind_conf->name);
  630. }
  631. rspamd_hard_terminate (rspamd_main);
  632. g_assert_not_reached ();
  633. }
  634. }
  635. else {
  636. spawn_worker_type (rspamd_main, ev_base, cf);
  637. }
  638. }
  639. cur = g_list_next (cur);
  640. }
  641. for (cw = workers; *cw != NULL; cw ++) {
  642. gboolean seen = FALSE;
  643. wrk = *cw;
  644. if (wrk->flags & RSPAMD_WORKER_ALWAYS_START) {
  645. for (i = 0; i < seen_mandatory_workers->len; i ++) {
  646. if (wrk == g_ptr_array_index (seen_mandatory_workers, i)) {
  647. seen = TRUE;
  648. break;
  649. }
  650. }
  651. if (!seen) {
  652. cf = rspamd_config_new_worker (rspamd_main->cfg, NULL);
  653. cf->count = 1;
  654. cf->worker = wrk;
  655. cf->type = g_quark_from_static_string (wrk->name);
  656. if (cf->worker->worker_init_func) {
  657. cf->ctx = cf->worker->worker_init_func (rspamd_main->cfg);
  658. }
  659. spawn_worker_type (rspamd_main, ev_base, cf);
  660. }
  661. }
  662. }
  663. g_ptr_array_free (seen_mandatory_workers, TRUE);
  664. }
  665. static void
  666. kill_old_workers (gpointer key, gpointer value, gpointer unused)
  667. {
  668. struct rspamd_worker *w = value;
  669. struct rspamd_main *rspamd_main;
  670. rspamd_main = w->srv;
  671. if (w->state == rspamd_worker_state_running) {
  672. w->state = rspamd_worker_state_terminating;
  673. kill (w->pid, SIGUSR2);
  674. ev_io_stop (rspamd_main->event_loop, &w->srv_ev);
  675. g_hash_table_remove_all (w->control_events_pending);
  676. msg_info_main ("send signal to worker %P", w->pid);
  677. }
  678. else {
  679. msg_info_main ("do not send signal to worker %P, already sent", w->pid);
  680. }
  681. }
  682. static void
  683. rspamd_worker_wait (struct rspamd_worker *w)
  684. {
  685. struct rspamd_main *rspamd_main;
  686. rspamd_main = w->srv;
  687. if (term_attempts < 0) {
  688. if (w->cf->worker->flags & RSPAMD_WORKER_KILLABLE) {
  689. if (kill (w->pid, SIGKILL) == -1) {
  690. if (errno == ESRCH) {
  691. /* We have actually killed the process */
  692. return;
  693. }
  694. }
  695. else {
  696. msg_warn_main ("terminate worker %s(%P) with SIGKILL",
  697. g_quark_to_string (w->type), w->pid);
  698. }
  699. }
  700. else {
  701. kill (w->pid, SIGKILL);
  702. if (errno == ESRCH) {
  703. /* We have actually killed the process */
  704. return;
  705. }
  706. else {
  707. msg_err_main ("data corruption warning: terminating "
  708. "special worker %s(%P) with SIGKILL",
  709. g_quark_to_string (w->type), w->pid);
  710. }
  711. }
  712. }
  713. }
  714. static void
  715. hash_worker_wait_callback (gpointer key, gpointer value, gpointer unused)
  716. {
  717. rspamd_worker_wait ((struct rspamd_worker *)value);
  718. }
  719. struct core_check_cbdata {
  720. struct rspamd_config *cfg;
  721. gsize total_count;
  722. gsize total_size;
  723. };
  724. #ifdef HAVE_NFTW
  725. static struct core_check_cbdata cores_cbdata;
  726. static gint
  727. rspamd_check_core_cb (const gchar *path, const struct stat *st,
  728. gint flag, struct FTW *ft)
  729. {
  730. if (S_ISREG (st->st_mode)) {
  731. cores_cbdata.total_count ++;
  732. /* Use physical size instead of displayed one */
  733. cores_cbdata.total_size += st->st_blocks * 512;
  734. }
  735. return 0;
  736. }
  737. #endif
  738. static void
  739. rspamd_check_core_limits (struct rspamd_main *rspamd_main)
  740. {
  741. #ifdef HAVE_NFTW
  742. struct rspamd_config *cfg = rspamd_main->cfg;
  743. cores_cbdata.cfg = cfg;
  744. cores_cbdata.total_count = 0;
  745. cores_cbdata.total_size = 0;
  746. if (cfg->cores_dir && (cfg->max_cores_count || cfg->max_cores_size)) {
  747. if (nftw (cfg->cores_dir, rspamd_check_core_cb, 1, FTW_MOUNT|FTW_PHYS)
  748. == -1) {
  749. msg_err_main ("nftw failed for path %s: %s", cfg->cores_dir,
  750. strerror (errno));
  751. }
  752. else {
  753. if (!rspamd_main->cores_throttling) {
  754. if (cfg->max_cores_size &&
  755. cores_cbdata.total_size > cfg->max_cores_size) {
  756. msg_warn_main (
  757. "enable cores throttling as size of cores in"
  758. " %s is %Hz, limit is %Hz",
  759. cfg->cores_dir,
  760. cores_cbdata.total_size,
  761. cfg->max_cores_size);
  762. rspamd_main->cores_throttling = TRUE;
  763. }
  764. if (cfg->max_cores_count &&
  765. cores_cbdata.total_count > cfg->max_cores_count) {
  766. msg_warn_main (
  767. "enable cores throttling as count of cores in"
  768. " %s is %z, limit is %z",
  769. cfg->cores_dir,
  770. cores_cbdata.total_count,
  771. cfg->max_cores_count);
  772. rspamd_main->cores_throttling = TRUE;
  773. }
  774. }
  775. else {
  776. if (cfg->max_cores_size &&
  777. cores_cbdata.total_size < cfg->max_cores_size) {
  778. msg_info_main (
  779. "disable cores throttling as size of cores in"
  780. " %s is now %Hz, limit is %Hz",
  781. cfg->cores_dir,
  782. cores_cbdata.total_size,
  783. cfg->max_cores_size);
  784. rspamd_main->cores_throttling = FALSE;
  785. }
  786. if (cfg->max_cores_count &&
  787. cores_cbdata.total_count < cfg->max_cores_count) {
  788. msg_info_main (
  789. "disable cores throttling as count of cores in"
  790. " %s is %z, limit is %z",
  791. cfg->cores_dir,
  792. cores_cbdata.total_count,
  793. cfg->max_cores_count);
  794. rspamd_main->cores_throttling = FALSE;
  795. }
  796. }
  797. }
  798. }
  799. #endif
  800. }
  801. static void
  802. reopen_log_handler (gpointer key, gpointer value, gpointer unused)
  803. {
  804. struct rspamd_worker *w = value;
  805. struct rspamd_main *rspamd_main;
  806. rspamd_main = w->srv;
  807. if (kill (w->pid, SIGUSR1) == -1) {
  808. msg_err_main ("kill failed for pid %P: %s", w->pid, strerror (errno));
  809. }
  810. }
  811. static gboolean
  812. load_rspamd_config (struct rspamd_main *rspamd_main,
  813. struct rspamd_config *cfg, gboolean init_modules,
  814. enum rspamd_post_load_options opts,
  815. gboolean reload)
  816. {
  817. cfg->compiled_modules = modules;
  818. cfg->compiled_workers = workers;
  819. if (!rspamd_config_read (cfg, cfg->cfg_name, config_logger, rspamd_main,
  820. ucl_vars, skip_template, lua_env)) {
  821. return FALSE;
  822. }
  823. /* Strictly set temp dir */
  824. if (!cfg->temp_dir) {
  825. msg_warn_main ("tempdir is not set, trying to use $TMPDIR");
  826. cfg->temp_dir =
  827. rspamd_mempool_strdup (cfg->cfg_pool, getenv ("TMPDIR"));
  828. if (!cfg->temp_dir) {
  829. msg_warn_main ("$TMPDIR is empty too, using /tmp as default");
  830. cfg->temp_dir = rspamd_mempool_strdup (cfg->cfg_pool, "/tmp");
  831. }
  832. }
  833. if (!reload) {
  834. /*
  835. * As some rules are defined in lua, we need to process them, then init
  836. * modules and merely afterwards to init modules
  837. */
  838. rspamd_lua_post_load_config (cfg);
  839. if (init_modules) {
  840. rspamd_init_filters (cfg, reload, false);
  841. }
  842. /* Do post-load actions */
  843. rspamd_config_post_load (cfg, opts);
  844. }
  845. return TRUE;
  846. }
  847. static void
  848. rspamd_detach_worker (struct rspamd_main *rspamd_main, struct rspamd_worker *wrk)
  849. {
  850. ev_io_stop (rspamd_main->event_loop, &wrk->srv_ev);
  851. ev_timer_stop (rspamd_main->event_loop, &wrk->hb.heartbeat_ev);
  852. }
  853. static void
  854. rspamd_attach_worker (struct rspamd_main *rspamd_main, struct rspamd_worker *wrk)
  855. {
  856. ev_io_start (rspamd_main->event_loop, &wrk->srv_ev);
  857. ev_timer_start (rspamd_main->event_loop, &wrk->hb.heartbeat_ev);
  858. }
  859. static void
  860. stop_srv_ev (gpointer key, gpointer value, gpointer ud)
  861. {
  862. struct rspamd_worker *cur = (struct rspamd_worker *)value;
  863. struct rspamd_main *rspamd_main = (struct rspamd_main *)ud;
  864. rspamd_detach_worker (rspamd_main, cur);
  865. }
  866. static void
  867. start_srv_ev (gpointer key, gpointer value, gpointer ud)
  868. {
  869. struct rspamd_worker *cur = (struct rspamd_worker *)value;
  870. struct rspamd_main *rspamd_main = (struct rspamd_main *)ud;
  871. rspamd_attach_worker (rspamd_main, cur);
  872. }
  873. static void
  874. rspamd_final_timer_handler (EV_P_ ev_timer *w, int revents)
  875. {
  876. struct rspamd_main *rspamd_main = (struct rspamd_main *)w->data;
  877. term_attempts--;
  878. g_hash_table_foreach (rspamd_main->workers, hash_worker_wait_callback,
  879. NULL);
  880. if (g_hash_table_size (rspamd_main->workers) == 0) {
  881. ev_break (rspamd_main->event_loop, EVBREAK_ALL);
  882. }
  883. }
  884. /* Signal handlers */
  885. static void
  886. rspamd_term_handler (struct ev_loop *loop, ev_signal *w, int revents)
  887. {
  888. struct rspamd_main *rspamd_main = (struct rspamd_main *)w->data;
  889. static ev_timer ev_finale;
  890. ev_tstamp shutdown_ts;
  891. if (!rspamd_main->wanna_die) {
  892. rspamd_main->wanna_die = TRUE;
  893. shutdown_ts = MAX (SOFT_SHUTDOWN_TIME,
  894. rspamd_main->cfg->task_timeout * 2.0);
  895. msg_info_main ("catch termination signal, waiting for %d children for %.2f seconds",
  896. (gint)g_hash_table_size (rspamd_main->workers),
  897. valgrind_mode ? shutdown_ts * 10 : shutdown_ts);
  898. /* Stop srv events to avoid false notifications */
  899. g_hash_table_foreach (rspamd_main->workers, stop_srv_ev, rspamd_main);
  900. rspamd_pass_signal (rspamd_main->workers, SIGTERM);
  901. if (control_fd != -1) {
  902. ev_io_stop (rspamd_main->event_loop, &control_ev);
  903. close (control_fd);
  904. }
  905. if (valgrind_mode) {
  906. /* Special case if we are likely running with valgrind */
  907. term_attempts = shutdown_ts / TERMINATION_INTERVAL * 10;
  908. }
  909. else {
  910. term_attempts = shutdown_ts / TERMINATION_INTERVAL;
  911. }
  912. ev_finale.data = rspamd_main;
  913. ev_timer_init (&ev_finale, rspamd_final_timer_handler,
  914. TERMINATION_INTERVAL, TERMINATION_INTERVAL);
  915. ev_timer_start (rspamd_main->event_loop, &ev_finale);
  916. }
  917. }
  918. static void
  919. rspamd_usr1_handler (struct ev_loop *loop, ev_signal *w, int revents)
  920. {
  921. struct rspamd_main *rspamd_main = (struct rspamd_main *)w->data;
  922. if (!rspamd_main->wanna_die) {
  923. rspamd_log_reopen (rspamd_main->logger,
  924. rspamd_main->cfg,
  925. rspamd_main->workers_uid,
  926. rspamd_main->workers_gid);
  927. msg_info_main ("logging reinitialised");
  928. g_hash_table_foreach (rspamd_main->workers, reopen_log_handler,
  929. NULL);
  930. }
  931. }
  932. static void
  933. rspamd_stat_update_handler (struct ev_loop *loop, ev_timer *w, int revents)
  934. {
  935. struct rspamd_main *rspamd_main = (struct rspamd_main *)w->data;
  936. struct rspamd_stat cur_stat;
  937. gchar proctitle[128];
  938. memcpy (&cur_stat, rspamd_main->stat, sizeof (cur_stat));
  939. if (old_stat.messages_scanned > 0 &&
  940. cur_stat.messages_scanned > old_stat.messages_scanned) {
  941. gdouble rate = (double)(cur_stat.messages_scanned - old_stat.messages_scanned) /
  942. w->repeat;
  943. gdouble old_spam = old_stat.actions_stat[METRIC_ACTION_REJECT] +
  944. old_stat.actions_stat[METRIC_ACTION_ADD_HEADER] +
  945. old_stat.actions_stat[METRIC_ACTION_REWRITE_SUBJECT];
  946. gdouble old_ham = old_stat.actions_stat[METRIC_ACTION_NOACTION];
  947. gdouble new_spam = cur_stat.actions_stat[METRIC_ACTION_REJECT] +
  948. cur_stat.actions_stat[METRIC_ACTION_ADD_HEADER] +
  949. cur_stat.actions_stat[METRIC_ACTION_REWRITE_SUBJECT];
  950. gdouble new_ham = cur_stat.actions_stat[METRIC_ACTION_NOACTION];
  951. rspamd_snprintf (proctitle, sizeof (proctitle),
  952. "main process; %.1f msg/sec, %.1f msg/sec spam, %.1f msg/sec ham",
  953. rate,
  954. (new_spam - old_spam) / w->repeat,
  955. (new_ham - old_ham) / w->repeat);
  956. setproctitle (proctitle);
  957. }
  958. memcpy (&old_stat, &cur_stat, sizeof (cur_stat));
  959. }
  960. static void
  961. rspamd_hup_handler (struct ev_loop *loop, ev_signal *w, int revents)
  962. {
  963. struct rspamd_main *rspamd_main = (struct rspamd_main *)w->data;
  964. if (!rspamd_main->wanna_die) {
  965. msg_info_main ("rspamd "
  966. RVERSION
  967. " is requested to reload configuration");
  968. /* Detach existing workers and stop their heartbeats */
  969. g_hash_table_foreach (rspamd_main->workers, stop_srv_ev, rspamd_main);
  970. if (reread_config (rspamd_main)) {
  971. msg_info_main ("kill old workers");
  972. g_hash_table_foreach (rspamd_main->workers, kill_old_workers, NULL);
  973. rspamd_check_core_limits (rspamd_main);
  974. msg_info_main ("spawn workers with a new config");
  975. spawn_workers (rspamd_main, rspamd_main->event_loop);
  976. msg_info_main ("workers spawning has been finished");
  977. }
  978. else {
  979. /* Reattach old workers */
  980. msg_info_main ("restore old workers with a old config");
  981. g_hash_table_foreach (rspamd_main->workers, start_srv_ev, rspamd_main);
  982. }
  983. }
  984. }
  985. /* Called when a dead child has been found */
  986. static void
  987. rspamd_cld_handler (EV_P_ ev_child *w, struct rspamd_main *rspamd_main,
  988. struct rspamd_worker *wrk)
  989. {
  990. gboolean need_refork;
  991. static struct rspamd_control_command cmd;
  992. /* Turn off locking for logger */
  993. ev_child_stop (EV_A_ w);
  994. /* Remove dead child form children list */
  995. g_hash_table_remove (rspamd_main->workers, GSIZE_TO_POINTER (wrk->pid));
  996. g_hash_table_remove_all (wrk->control_events_pending);
  997. if (wrk->srv_pipe[0] != -1) {
  998. /* Ugly workaround */
  999. if (wrk->tmp_data) {
  1000. g_free (wrk->tmp_data);
  1001. }
  1002. rspamd_detach_worker (rspamd_main, wrk);
  1003. }
  1004. if (wrk->control_pipe[0] != -1) {
  1005. /* We also need to clean descriptors left */
  1006. close (wrk->control_pipe[0]);
  1007. close (wrk->srv_pipe[0]);
  1008. }
  1009. if (!rspamd_main->wanna_die) {
  1010. cmd.type = RSPAMD_CONTROL_CHILD_CHANGE;
  1011. cmd.cmd.child_change.what = rspamd_child_terminated;
  1012. cmd.cmd.child_change.pid = wrk->pid;
  1013. cmd.cmd.child_change.additional = w->rstatus;
  1014. rspamd_control_broadcast_srv_cmd (rspamd_main, &cmd, wrk->pid);
  1015. }
  1016. need_refork = rspamd_check_termination_clause (wrk->srv, wrk, w->rstatus);
  1017. if (need_refork) {
  1018. /* Fork another worker in replace of dead one */
  1019. msg_info_main ("respawn process %s in lieu of terminated process with pid %P",
  1020. g_quark_to_string (wrk->type),
  1021. wrk->pid);
  1022. rspamd_check_core_limits (rspamd_main);
  1023. rspamd_fork_delayed (wrk->cf, wrk->index, rspamd_main);
  1024. }
  1025. else {
  1026. msg_info_main ("do not respawn process %s after found terminated process with pid %P",
  1027. g_quark_to_string (wrk->type),
  1028. wrk->pid);
  1029. }
  1030. REF_RELEASE (wrk->cf);
  1031. g_hash_table_unref (wrk->control_events_pending);
  1032. g_free (wrk);
  1033. }
  1034. /* Control socket handler */
  1035. static void
  1036. rspamd_control_handler (EV_P_ ev_io *w, int revents)
  1037. {
  1038. struct rspamd_main *rspamd_main = (struct rspamd_main *)w->data;
  1039. rspamd_inet_addr_t *addr;
  1040. gint nfd;
  1041. if ((nfd =
  1042. rspamd_accept_from_socket (w->fd, &addr, NULL, NULL)) == -1) {
  1043. msg_warn_main ("accept failed: %s", strerror (errno));
  1044. return;
  1045. }
  1046. /* Check for EAGAIN */
  1047. if (nfd == 0) {
  1048. return;
  1049. }
  1050. msg_info_main ("accepted control connection from %s",
  1051. rspamd_inet_address_to_string (addr));
  1052. rspamd_control_process_client_socket (rspamd_main, nfd, addr);
  1053. }
  1054. static guint
  1055. rspamd_spair_hash (gconstpointer p)
  1056. {
  1057. return rspamd_cryptobox_fast_hash (p, PAIR_ID_LEN, rspamd_hash_seed ());
  1058. }
  1059. static gboolean
  1060. rspamd_spair_equal (gconstpointer a, gconstpointer b)
  1061. {
  1062. return memcmp (a, b, PAIR_ID_LEN) == 0;
  1063. }
  1064. static void
  1065. rspamd_spair_close (gpointer p)
  1066. {
  1067. gint *fds = p;
  1068. close (fds[0]);
  1069. close (fds[1]);
  1070. g_free (p);
  1071. }
  1072. static void
  1073. version (void)
  1074. {
  1075. #if defined(GIT_VERSION) && GIT_VERSION == 1
  1076. rspamd_printf ("Rspamd daemon version " RVERSION "-git." RID "\n");
  1077. #else
  1078. rspamd_printf ("Rspamd daemon version " RVERSION "\n");
  1079. #endif
  1080. }
  1081. gint
  1082. main (gint argc, gchar **argv, gchar **env)
  1083. {
  1084. gint i, res = 0;
  1085. struct sigaction signals, sigpipe_act;
  1086. worker_t **pworker;
  1087. GQuark type;
  1088. rspamd_inet_addr_t *control_addr = NULL;
  1089. struct ev_loop *event_loop;
  1090. struct rspamd_main *rspamd_main;
  1091. gboolean skip_pid = FALSE;
  1092. rspamd_main = (struct rspamd_main *) g_malloc0 (sizeof (struct rspamd_main));
  1093. rspamd_main->server_pool = rspamd_mempool_new (rspamd_mempool_suggest_size (),
  1094. "main", 0);
  1095. rspamd_main->stat = rspamd_mempool_alloc0_shared (rspamd_main->server_pool,
  1096. sizeof (struct rspamd_stat));
  1097. rspamd_main->cfg = rspamd_config_new (RSPAMD_CONFIG_INIT_DEFAULT);
  1098. rspamd_main->spairs = g_hash_table_new_full (rspamd_spair_hash,
  1099. rspamd_spair_equal, g_free, rspamd_spair_close);
  1100. rspamd_main->start_mtx = rspamd_mempool_get_mutex (rspamd_main->server_pool);
  1101. if (getenv ("VALGRIND") != NULL) {
  1102. valgrind_mode = TRUE;
  1103. }
  1104. #ifndef HAVE_SETPROCTITLE
  1105. init_title (rspamd_main->server_pool, argc, argv, env);
  1106. #endif
  1107. rspamd_main->cfg->libs_ctx = rspamd_init_libs ();
  1108. memset (&signals, 0, sizeof (struct sigaction));
  1109. read_cmd_line (&argc, &argv, rspamd_main->cfg);
  1110. if (show_version) {
  1111. version ();
  1112. exit (EXIT_SUCCESS);
  1113. }
  1114. if (argc > 0) {
  1115. /* Parse variables */
  1116. for (i = 0; i < argc; i++) {
  1117. if (strchr (argv[i], '=') != NULL) {
  1118. gchar *k, *v, *t;
  1119. k = g_strdup (argv[i]);
  1120. t = strchr (k, '=');
  1121. v = g_strdup (t + 1);
  1122. *t = '\0';
  1123. if (ucl_vars == NULL) {
  1124. ucl_vars = g_hash_table_new_full (rspamd_strcase_hash,
  1125. rspamd_strcase_equal, g_free, g_free);
  1126. }
  1127. g_hash_table_insert (ucl_vars, k, v);
  1128. }
  1129. }
  1130. }
  1131. if (is_debug) {
  1132. rspamd_main->cfg->log_level = G_LOG_LEVEL_DEBUG;
  1133. }
  1134. else {
  1135. rspamd_main->cfg->log_level = G_LOG_LEVEL_MESSAGE;
  1136. }
  1137. type = g_quark_from_static_string ("main");
  1138. /* First set logger to console logger */
  1139. rspamd_main->logger = rspamd_log_open_emergency (rspamd_main->server_pool);
  1140. g_assert (rspamd_main->logger != NULL);
  1141. if (is_debug) {
  1142. rspamd_log_set_log_level (rspamd_main->logger, G_LOG_LEVEL_DEBUG);
  1143. }
  1144. else {
  1145. rspamd_log_set_log_level (rspamd_main->logger, G_LOG_LEVEL_MESSAGE);
  1146. }
  1147. g_log_set_default_handler (rspamd_glib_log_function, rspamd_main->logger);
  1148. g_set_printerr_handler (rspamd_glib_printerr_function);
  1149. detect_priv (rspamd_main);
  1150. pworker = &workers[0];
  1151. while (*pworker) {
  1152. /* Init string quarks */
  1153. (void) g_quark_from_static_string ((*pworker)->name);
  1154. pworker++;
  1155. }
  1156. /* Init listen sockets hash */
  1157. listen_sockets = g_hash_table_new (g_direct_hash, g_direct_equal);
  1158. sqlite3_initialize ();
  1159. /* Load config */
  1160. if (!load_rspamd_config (rspamd_main, rspamd_main->cfg, TRUE,
  1161. RSPAMD_CONFIG_LOAD_ALL, FALSE)) {
  1162. exit (EXIT_FAILURE);
  1163. }
  1164. /* Override pidfile from configuration by command line argument */
  1165. if (rspamd_pidfile != NULL) {
  1166. rspamd_main->cfg->pid_file = rspamd_pidfile;
  1167. }
  1168. /* Force debug log */
  1169. if (is_debug) {
  1170. rspamd_log_set_log_level (rspamd_main->logger, G_LOG_LEVEL_DEBUG);
  1171. }
  1172. /* Create rolling history */
  1173. rspamd_main->history = rspamd_roll_history_new (rspamd_main->server_pool,
  1174. rspamd_main->cfg->history_rows, rspamd_main->cfg);
  1175. msg_info_main ("rspamd "
  1176. RVERSION
  1177. " is starting, build id: "
  1178. RID);
  1179. rspamd_main->cfg->cfg_name = rspamd_mempool_strdup (
  1180. rspamd_main->cfg->cfg_pool,
  1181. rspamd_main->cfg->cfg_name);
  1182. msg_info_main ("cpu features: %s",
  1183. rspamd_main->cfg->libs_ctx->crypto_ctx->cpu_extensions);
  1184. msg_info_main ("cryptobox configuration: curve25519(libsodium), "
  1185. "chacha20(%s), poly1305(libsodium), siphash(libsodium), blake2(libsodium), base64(%s)",
  1186. rspamd_main->cfg->libs_ctx->crypto_ctx->chacha20_impl,
  1187. rspamd_main->cfg->libs_ctx->crypto_ctx->base64_impl);
  1188. msg_info_main ("libottery prf: %s", ottery_get_impl_name ());
  1189. /* Daemonize */
  1190. if (!no_fork) {
  1191. if (daemon (0, 0) == -1) {
  1192. msg_err_main ("cannot daemonize: %s", strerror (errno));
  1193. exit (-errno);
  1194. }
  1195. /* Close emergency logger */
  1196. rspamd_log_close (rspamd_log_emergency_logger ());
  1197. }
  1198. /* Write info */
  1199. rspamd_main->pid = getpid ();
  1200. rspamd_main->type = type;
  1201. if (!valgrind_mode) {
  1202. rspamd_set_crash_handler (rspamd_main);
  1203. }
  1204. /* Ignore SIGPIPE as we handle write errors manually */
  1205. sigemptyset (&sigpipe_act.sa_mask);
  1206. sigaddset (&sigpipe_act.sa_mask, SIGPIPE);
  1207. sigpipe_act.sa_handler = SIG_IGN;
  1208. sigpipe_act.sa_flags = 0;
  1209. sigaction (SIGPIPE, &sigpipe_act, NULL);
  1210. if (rspamd_main->cfg->pid_file == NULL) {
  1211. msg_info_main ("pid file is not specified, skipping writing it");
  1212. skip_pid = TRUE;
  1213. }
  1214. else if (no_fork) {
  1215. msg_info_main ("skip writing pid in no-fork mode");
  1216. skip_pid = TRUE;
  1217. }
  1218. else if (rspamd_write_pid (rspamd_main) == -1) {
  1219. msg_err_main ("cannot write pid file %s", rspamd_main->cfg->pid_file);
  1220. exit (-errno);
  1221. }
  1222. /* Block signals to use sigsuspend in future */
  1223. sigprocmask (SIG_BLOCK, &signals.sa_mask, NULL);
  1224. /* Set title */
  1225. setproctitle ("main process");
  1226. /* Open control socket if needed */
  1227. control_fd = -1;
  1228. if (rspamd_main->cfg->control_socket_path) {
  1229. if (!rspamd_parse_inet_address (&control_addr,
  1230. rspamd_main->cfg->control_socket_path,
  1231. strlen (rspamd_main->cfg->control_socket_path),
  1232. RSPAMD_INET_ADDRESS_PARSE_DEFAULT)) {
  1233. msg_err_main ("cannot parse inet address %s",
  1234. rspamd_main->cfg->control_socket_path);
  1235. }
  1236. else {
  1237. control_fd = rspamd_inet_address_listen (control_addr, SOCK_STREAM,
  1238. RSPAMD_INET_ADDRESS_LISTEN_ASYNC, -1);
  1239. if (control_fd == -1) {
  1240. msg_err_main ("cannot open control socket at path: %s",
  1241. rspamd_main->cfg->control_socket_path);
  1242. }
  1243. }
  1244. }
  1245. /* Maybe read roll history */
  1246. if (rspamd_main->cfg->history_file) {
  1247. rspamd_roll_history_load (rspamd_main->history,
  1248. rspamd_main->cfg->history_file);
  1249. }
  1250. /* Spawn workers */
  1251. rspamd_main->workers = g_hash_table_new (g_direct_hash, g_direct_equal);
  1252. /* Init event base */
  1253. event_loop = ev_default_loop (rspamd_config_ev_backend_get (rspamd_main->cfg));
  1254. rspamd_main->event_loop = event_loop;
  1255. if (event_loop) {
  1256. int loop_type = ev_backend (event_loop);
  1257. gboolean effective_backend;
  1258. const gchar *loop_str;
  1259. loop_str =
  1260. rspamd_config_ev_backend_to_string (loop_type, &effective_backend);
  1261. if (!effective_backend) {
  1262. msg_warn_main ("event loop uses non-optimal backend: %s", loop_str);
  1263. }
  1264. else {
  1265. msg_info_main ("event loop initialised with backend: %s", loop_str);
  1266. }
  1267. }
  1268. else {
  1269. msg_err ("cannot init event loop! exiting");
  1270. exit (EXIT_FAILURE);
  1271. }
  1272. /* Unblock signals */
  1273. sigemptyset (&signals.sa_mask);
  1274. sigprocmask (SIG_SETMASK, &signals.sa_mask, NULL);
  1275. /* Set events for signals */
  1276. ev_signal_init (&rspamd_main->term_ev, rspamd_term_handler, SIGTERM);
  1277. rspamd_main->term_ev.data = rspamd_main;
  1278. ev_signal_start (event_loop, &rspamd_main->term_ev);
  1279. ev_signal_init (&rspamd_main->int_ev, rspamd_term_handler, SIGINT);
  1280. rspamd_main->int_ev.data = rspamd_main;
  1281. ev_signal_start (event_loop, &rspamd_main->int_ev);
  1282. ev_signal_init (&rspamd_main->hup_ev, rspamd_hup_handler, SIGHUP);
  1283. rspamd_main->hup_ev.data = rspamd_main;
  1284. ev_signal_start (event_loop, &rspamd_main->hup_ev);
  1285. ev_signal_init (&rspamd_main->usr1_ev, rspamd_usr1_handler, SIGUSR1);
  1286. rspamd_main->usr1_ev.data = rspamd_main;
  1287. ev_signal_start (event_loop, &rspamd_main->usr1_ev);
  1288. /* Update proctitle according to number of messages processed */
  1289. static const ev_tstamp stat_update_time = 10.0;
  1290. memset (&old_stat, 0, sizeof (old_stat));
  1291. stat_ev.data = rspamd_main;
  1292. ev_timer_init (&stat_ev, rspamd_stat_update_handler,
  1293. stat_update_time, stat_update_time);
  1294. ev_timer_start (event_loop, &stat_ev);
  1295. rspamd_check_core_limits (rspamd_main);
  1296. rspamd_mempool_lock_mutex (rspamd_main->start_mtx);
  1297. spawn_workers (rspamd_main, event_loop);
  1298. rspamd_mempool_unlock_mutex (rspamd_main->start_mtx);
  1299. rspamd_main->http_ctx = rspamd_http_context_create (rspamd_main->cfg,
  1300. event_loop, rspamd_main->cfg->ups_ctx);
  1301. if (control_fd != -1) {
  1302. msg_info_main ("listening for control commands on %s",
  1303. rspamd_inet_address_to_string (control_addr));
  1304. ev_io_init (&control_ev, rspamd_control_handler, control_fd, EV_READ);
  1305. control_ev.data = rspamd_main;
  1306. ev_io_start (event_loop, &control_ev);
  1307. }
  1308. ev_loop (event_loop, 0);
  1309. /* Maybe save roll history */
  1310. if (rspamd_main->cfg->history_file) {
  1311. rspamd_roll_history_save (rspamd_main->history,
  1312. rspamd_main->cfg->history_file);
  1313. }
  1314. msg_info_main ("terminating...");
  1315. REF_RELEASE (rspamd_main->cfg);
  1316. rspamd_log_close (rspamd_main->logger);
  1317. g_hash_table_unref (rspamd_main->spairs);
  1318. g_hash_table_unref (rspamd_main->workers);
  1319. rspamd_mempool_delete (rspamd_main->server_pool);
  1320. if (!skip_pid) {
  1321. rspamd_pidfile_close (rspamd_main->pfh);
  1322. }
  1323. g_free (rspamd_main);
  1324. ev_unref (event_loop);
  1325. sqlite3_shutdown ();
  1326. if (control_addr) {
  1327. rspamd_inet_address_free (control_addr);
  1328. }
  1329. return (res);
  1330. }