#include "statfile_sync.h"
#include "lua/lua_common.h"
#include "dynamic_cfg.h"
+#include "rrd.h"
#define END "END" CRLF
/* 120 seconds for controller's IO */
#define CONTROLLER_IO_TIMEOUT 120
+/* RRD macroes */
+/* Write data each minute */
+#define CONTROLLER_RRD_STEP 60
+
/* Init functions */
gpointer init_controller (void);
void start_controller (struct rspamd_worker *worker);
guint32 timeout;
struct rspamd_dns_resolver *resolver;
struct event_base *ev_base;
+ struct event rrd_event;
+ struct rspamd_rrd_file *rrd_file;
+ struct rspamd_main *srv;
};
static struct controller_command commands[] = {
#endif
}
+static gboolean
+create_rrd_file (const gchar *filename, struct rspamd_controller_ctx *ctx)
+{
+ GError *err = NULL;
+ GArray ar;
+ struct rrd_rra_def rra[5];
+ struct rrd_ds_def ds[4];
+
+ /*
+ * DS:
+ * 1) reject as spam
+ * 2) mark as spam
+ * 3) greylist
+ * 4) pass
+ */
+ /*
+ * RRA:
+ * 1) per minute AVERAGE
+ * 2) per 5 minutes AVERAGE
+ * 3) per 30 minutes AVERAGE
+ * 4) per 2 hours AVERAGE
+ * 5) per day AVERAGE
+ */
+ ctx->rrd_file = rspamd_rrd_create (filename, 4, 5, CONTROLLER_RRD_STEP, &err);
+ if (ctx->rrd_file == NULL) {
+ msg_err ("cannot create rrd file %s, error: %s", filename, err->message);
+ g_error_free (err);
+ return FALSE;
+ }
+
+ /* Add all ds and rra */
+ rrd_make_default_ds ("spam", CONTROLLER_RRD_STEP, &ds[0]);
+ rrd_make_default_ds ("possible spam", CONTROLLER_RRD_STEP, &ds[1]);
+ rrd_make_default_ds ("greylist", CONTROLLER_RRD_STEP, &ds[2]);
+ rrd_make_default_ds ("ham", CONTROLLER_RRD_STEP, &ds[3]);
+
+ rrd_make_default_rra ("AVERAGE", 1, 600, &rra[0]);
+ rrd_make_default_rra ("AVERAGE", 5, 600, &rra[1]);
+ rrd_make_default_rra ("AVERAGE", 30, 700, &rra[2]);
+ rrd_make_default_rra ("AVERAGE", 120, 775, &rra[3]);
+ rrd_make_default_rra ("AVERAGE", 1440, 797, &rra[4]);
+
+ ar.data = (gchar *)ds;
+ ar.len = sizeof (ds);
+ if (!rspamd_rrd_add_ds (ctx->rrd_file, &ar, &err)) {
+ msg_err ("cannot create rrd file %s, error: %s", filename, err->message);
+ g_error_free (err);
+ rspamd_rrd_close (ctx->rrd_file);
+ return FALSE;
+ }
+
+ ar.data = (gchar *)rra;
+ ar.len = sizeof (rra);
+ if (!rspamd_rrd_add_rra (ctx->rrd_file, &ar, &err)) {
+ msg_err ("cannot create rrd file %s, error: %s", filename, err->message);
+ g_error_free (err);
+ rspamd_rrd_close (ctx->rrd_file);
+ return FALSE;
+ }
+
+ /* Finalize */
+ if (!rspamd_rrd_finalize (ctx->rrd_file, &err)) {
+ msg_err ("cannot create rrd file %s, error: %s", filename, err->message);
+ g_error_free (err);
+ rspamd_rrd_close (ctx->rrd_file);
+ return FALSE;
+ }
+
+ return TRUE;
+}
+
+static void
+controller_update_rrd (gint fd, short what, void *arg)
+{
+ struct rspamd_controller_ctx *ctx = arg;
+ struct timeval tv;
+ GArray ar;
+ gdouble data[4];
+ GError *err = NULL;
+
+ /*
+ * Data:
+ * 1) reject as spam
+ * 2) mark as spam
+ * 3) greylist
+ * 4) pass
+ */
+
+ tv.tv_sec = CONTROLLER_RRD_STEP;
+ tv.tv_usec = 0;
+
+ /* Fill data */
+ data[0] = ctx->srv->stat->actions_stat[METRIC_ACTION_REJECT];
+ data[1] = ctx->srv->stat->actions_stat[METRIC_ACTION_ADD_HEADER] + ctx->srv->stat->actions_stat[METRIC_ACTION_REWRITE_SUBJECT];
+ data[2] = ctx->srv->stat->actions_stat[METRIC_ACTION_GREYLIST];
+ data[3] = ctx->srv->stat->actions_stat[METRIC_ACTION_NOACTION];
+
+ ar.data = (gchar *)data;
+ ar.len = sizeof (data);
+ if (!rspamd_rrd_add_record (ctx->rrd_file, &ar, &err)) {
+ msg_err ("cannot add record to rrd database: %s, stop rrd update", err->message);
+ g_error_free (err);
+ }
+ else {
+ evtimer_add (&ctx->rrd_event, &tv);
+ }
+}
+
gpointer
init_controller (void)
{
gchar *hostbuf;
gsize hostmax;
struct rspamd_controller_ctx *ctx;
+ GError *err = NULL;
+ struct timeval tv;
worker->srv->pid = getpid ();
ctx = worker->ctx;
msg_info ("cannot start statfile synchronization, statfiles would not be synchronized");
}
+ /* Check for rrd */
+ tv.tv_sec = CONTROLLER_RRD_STEP;
+ tv.tv_usec = 0;
+ ctx->srv = worker->srv;
+ if (worker->srv->cfg->rrd_file) {
+ ctx->rrd_file = rspamd_rrd_open (worker->srv->cfg->rrd_file, &err);
+ if (ctx->rrd_file == NULL) {
+ msg_info ("cannot open rrd file: %s, error: %s, trying to create", worker->srv->cfg->rrd_file, err->message);
+ g_error_free (err);
+ /* Try to create rrd file */
+ if (create_rrd_file (worker->srv->cfg->rrd_file, ctx)) {
+ evtimer_set (&ctx->rrd_event, controller_update_rrd, ctx);
+ event_base_set (ctx->ev_base, &ctx->rrd_event);
+ evtimer_add (&ctx->rrd_event, &tv);
+ }
+ }
+ else {
+ evtimer_set (&ctx->rrd_event, controller_update_rrd, ctx);
+ event_base_set (ctx->ev_base, &ctx->rrd_event);
+ evtimer_add (&ctx->rrd_event, &tv);
+ }
+ }
+
/* Fill hostname buf */
hostmax = sysconf (_SC_HOST_NAME_MAX) + 1;
hostbuf = alloca (hostmax);
event_base_loop (ctx->ev_base, 0);
close_log (worker->srv->logger);
+ if (ctx->rrd_file) {
+ rspamd_rrd_close (ctx->rrd_file);
+ }
exit (EXIT_SUCCESS);
}
--- /dev/null
+/* Copyright (c) 2010-2012, Vsevolod Stakhov
+ * All rights reserved.
+ *
+ * Redistribution and use in source and binary forms, with or without
+ * modification, are permitted provided that the following conditions are met:
+ * * Redistributions of source code must retain the above copyright
+ * notice, this list of conditions and the following disclaimer.
+ * * Redistributions in binary form must reproduce the above copyright
+ * notice, this list of conditions and the following disclaimer in the
+ * documentation and/or other materials provided with the distribution.
+ *
+ * THIS SOFTWARE IS PROVIDED ''AS IS'' AND ANY
+ * EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
+ * WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
+ * DISCLAIMED. IN NO EVENT SHALL AUTHOR BE LIABLE FOR ANY
+ * DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES
+ * (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
+ * LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND
+ * ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+ * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
+ * SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ */
+
+
+#include "config.h"
+#include "util.h"
+#include "main.h"
+#include "message.h"
+#include "protocol.h"
+#include "upstream.h"
+#include "cfg_file.h"
+#include "cfg_xml.h"
+#include "map.h"
+#include "dns.h"
+#include "tokenizers/tokenizers.h"
+#include "classifiers/classifiers.h"
+#include "dynamic_cfg.h"
+#include "rrd.h"
+
+#include <evhttp.h>
+#if (_EVENT_NUMERIC_VERSION > 0x02010000) && defined(HAVE_OPENSSL)
+#define HAVE_WEBUI_SSL
+#include <openssl/ssl.h>
+#include <openssl/rand.h>
+#include <event2/event-config.h>
+#include <event2/bufferevent.h>
+#include <event2/util.h>
+#include <event2/bufferevent_ssl.h>
+#endif
+
+#ifdef WITH_GPERF_TOOLS
+# include <glib/gprintf.h>
+#endif
+
+/* 60 seconds for worker's IO */
+#define DEFAULT_WORKER_IO_TIMEOUT 60000
+
+/* HTTP paths */
+#define PATH_AUTH "/login"
+
+gpointer init_webui_worker (void);
+void start_webui_worker (struct rspamd_worker *worker);
+
+worker_t webui_worker = {
+ "webui", /* Name */
+ init_webui_worker, /* Init function */
+ start_webui_worker, /* Start function */
+ TRUE, /* Has socket */
+ TRUE, /* Non unique */
+ FALSE, /* Non threaded */
+ TRUE /* Killable */
+};
+
+/*
+ * Worker's context
+ */
+struct rspamd_webui_worker_ctx {
+ /* DNS resolver */
+ struct rspamd_dns_resolver *resolver;
+ /* Events base */
+ struct event_base *ev_base;
+ /* Whether we use ssl for this server */
+ gboolean use_ssl;
+ /* Webui password */
+ gchar *password;
+ /* HTTP server */
+ struct evhttp *http;
+ /* Server's start time */
+ time_t start_time;
+ /* Main server */
+ struct rspamd_main *srv;
+ /* Configuration */
+ struct config_file *cfg;
+ /* SSL cert */
+ gchar *ssl_cert;
+ /* SSL private key */
+ gchar *ssl_key;
+};
+
+static sig_atomic_t wanna_die = 0;
+
+/* Signal handlers */
+
+#ifndef HAVE_SA_SIGINFO
+static void
+sig_handler (gint signo)
+#else
+static void
+sig_handler (gint signo, siginfo_t * info, void *unused)
+#endif
+{
+ struct timeval tv;
+
+ switch (signo) {
+ case SIGINT:
+ case SIGTERM:
+ if (!wanna_die) {
+ wanna_die = 1;
+ tv.tv_sec = 0;
+ tv.tv_usec = 0;
+ event_loopexit (&tv);
+
+#ifdef WITH_GPERF_TOOLS
+ ProfilerStop ();
+#endif
+ }
+ break;
+ }
+}
+
+/*
+ * Config reload is designed by sending sigusr2 to active workers and pending shutdown of them
+ */
+static void
+sigusr2_handler (gint fd, short what, void *arg)
+{
+ struct rspamd_worker *worker = (struct rspamd_worker *) arg;
+ /* Do not accept new connections, preparing to end worker's process */
+ struct timeval tv;
+
+ if (!wanna_die) {
+ tv.tv_sec = SOFT_SHUTDOWN_TIME;
+ tv.tv_usec = 0;
+ event_del (&worker->sig_ev_usr1);
+ event_del (&worker->sig_ev_usr2);
+ event_del (&worker->bind_ev);
+ msg_info ("worker's shutdown is pending in %d sec", SOFT_SHUTDOWN_TIME);
+ event_loopexit (&tv);
+ }
+ return;
+}
+
+/*
+ * Reopen log is designed by sending sigusr1 to active workers and pending shutdown of them
+ */
+static void
+sigusr1_handler (gint fd, short what, void *arg)
+{
+ struct rspamd_worker *worker = (struct rspamd_worker *) arg;
+
+ reopen_log (worker->srv->logger);
+
+ return;
+}
+
+#ifdef HAVE_WEBUI_SSL
+
+static struct bufferevent*
+webui_ssl_bufferevent_gen (struct event_base *base, void *arg)
+{
+ SSL_CTX *server_ctx = arg;
+ SSL *client_ctx;
+ struct bufferevent *base_bev, *ssl_bev;
+
+ client_ctx = SSL_new (server_ctx);
+
+ base_bev = bufferevent_socket_new (base, -1, 0);
+ if (base_bev == NULL) {
+ msg_err ("cannot create base bufferevent for ssl connection");
+ return NULL;
+ }
+
+ ssl_bev = bufferevent_openssl_filter_new (base, base_bev, client_ctx, BUFFEREVENT_SSL_ACCEPTING, 0);
+
+ if (ssl_bev == NULL) {
+ msg_err ("cannot create ssl bufferevent for ssl connection");
+ }
+
+ return ssl_bev;
+}
+
+static void
+webui_ssl_init (struct rspamd_webui_worker_ctx *ctx)
+{
+ SSL_CTX *server_ctx;
+
+ /* Initialize the OpenSSL library */
+ SSL_load_error_strings ();
+ SSL_library_init ();
+ /* We MUST have entropy, or else there's no point to crypto. */
+ if (!RAND_poll ()) {
+ return NULL;
+ }
+
+ server_ctx = SSL_CTX_new (SSLv23_server_method ());
+
+ if (! SSL_CTX_use_certificate_chain_file (server_ctx, ctx->ssl_cert) ||
+ ! SSL_CTX_use_PrivateKey_file(server_ctx, ctx->ssl_key, SSL_FILETYPE_PEM)) {
+ msg_err ("cannot load ssl key %s or ssl cert: %s", ctx->ssl_key, ctx->ssl_cert);
+ return;
+ }
+ SSL_CTX_set_options (server_ctx, SSL_OP_NO_SSLv2);
+
+ if (server_ctx) {
+ /* Set generator for ssl events */
+ evhttp_set_bevcb (ctx->http, webui_ssl_bufferevent_gen, server_ctx);
+ }
+}
+#endif
+
+/* Command handlers */
+
+/*
+ * Auth command handler:
+ * request: /auth
+ * headers: Password
+ * reply: json {"auth": "ok", "version": "0.5.2", "uptime": "some uptime", "error": "none"}
+ */
+static void
+http_handle_auth (struct evhttp_request *req, gpointer arg)
+{
+ struct rspamd_webui_worker_ctx *ctx = arg;
+ struct evbuffer *evb;
+ const gchar *password;
+ gchar *auth = "ok", *error = "none", uptime_buf[128];
+ time_t uptime, days, hours, minutes;
+
+ evb = evbuffer_new ();
+ if (!evb) {
+ msg_err ("cannot allocate evbuffer for reply");
+ return;
+ }
+
+ if (ctx->password) {
+ password = evhttp_find_header (req->input_headers, "Password");
+ if (password == NULL || strcmp (password, ctx->password) != 0) {
+ auth = "failed";
+ error = "unauthorized";
+ }
+ }
+ /* Print uptime */
+ uptime = time (NULL) - ctx->start_time;
+ /* If uptime more than 2 hours, print as a number of days. */
+ if (uptime >= 2 * 3600) {
+ days = uptime / 86400;
+ hours = uptime / 3600 - days * 24;
+ minutes = uptime / 60 - hours * 60 - days * 1440;
+ rspamd_snprintf (uptime_buf, sizeof (uptime_buf), "%d day%s %d hour%s %d minute%s", days, days != 1 ? "s" : " ", hours, hours != 1 ? "s" : " ", minutes, minutes != 1 ? "s" : " ");
+ }
+ /* If uptime is less than 1 minute print only seconds */
+ else if (uptime / 60 == 0) {
+ rspamd_snprintf (uptime_buf, sizeof (uptime_buf), "%d second%s", (gint)uptime, (gint)uptime != 1 ? "s" : " ");
+ }
+ /* Else print the minutes and seconds. */
+ else {
+ hours = uptime / 3600;
+ minutes = uptime / 60 - hours * 60;
+ uptime -= hours * 3600 + minutes * 60;
+ rspamd_snprintf (uptime_buf, sizeof (uptime_buf), "%d hour%s %d minute%s %d second%s", hours, hours > 1 ? "s" : " ", minutes, minutes > 1 ? "s" : " ", (gint)uptime, uptime > 1 ? "s" : " ");
+ }
+
+ evbuffer_add_printf (evb, "{\"auth\": \"%s\", \"version\": \"%s\", \"uptime\": \"%s\", \"error\": \"%s\"}" CRLF,
+ auth, RVERSION, uptime_buf, error);
+ evhttp_add_header(req->output_headers, "Connection", "close");
+
+ evhttp_send_reply(req, HTTP_OK, "OK", evb);
+ evbuffer_free(evb);
+}
+
+gpointer
+init_webui_worker (void)
+{
+ struct rspamd_webui_worker_ctx *ctx;
+ GQuark type;
+
+ type = g_quark_try_string ("webui");
+
+ ctx = g_malloc0 (sizeof (struct rspamd_webui_worker_ctx));
+
+ register_worker_opt (type, "password", xml_handle_string, ctx, G_STRUCT_OFFSET (struct rspamd_webui_worker_ctx, password));
+ register_worker_opt (type, "ssl", xml_handle_boolean, ctx, G_STRUCT_OFFSET (struct rspamd_webui_worker_ctx, use_ssl));
+ register_worker_opt (type, "ssl_cert", xml_handle_string, ctx, G_STRUCT_OFFSET (struct rspamd_webui_worker_ctx, ssl_cert));
+ register_worker_opt (type, "ssl_key", xml_handle_string, ctx, G_STRUCT_OFFSET (struct rspamd_webui_worker_ctx, ssl_key));
+
+ return ctx;
+}
+
+/*
+ * Start worker process
+ */
+void
+start_webui_worker (struct rspamd_worker *worker)
+{
+ struct sigaction signals;
+ struct rspamd_webui_worker_ctx *ctx = worker->ctx;
+
+#ifdef WITH_PROFILER
+ extern void _start (void), etext (void);
+ monstartup ((u_long) & _start, (u_long) & etext);
+#endif
+
+ gperf_profiler_init (worker->srv->cfg, "webui_worker");
+
+ worker->srv->pid = getpid ();
+
+ ctx->ev_base = event_init ();
+
+ ctx->cfg = worker->srv->cfg;
+ ctx->srv = worker->srv;
+
+ init_signals (&signals, sig_handler);
+ sigprocmask (SIG_UNBLOCK, &signals.sa_mask, NULL);
+
+ /* SIGUSR2 handler */
+ signal_set (&worker->sig_ev_usr2, SIGUSR2, sigusr2_handler, (void *) worker);
+ event_base_set (ctx->ev_base, &worker->sig_ev_usr2);
+ signal_add (&worker->sig_ev_usr2, NULL);
+
+ /* SIGUSR1 handler */
+ signal_set (&worker->sig_ev_usr1, SIGUSR1, sigusr1_handler, (void *) worker);
+ event_base_set (ctx->ev_base, &worker->sig_ev_usr1);
+ signal_add (&worker->sig_ev_usr1, NULL);
+
+ ctx->start_time = time (NULL);
+ /* Accept event */
+ ctx->http = evhttp_new (ctx->ev_base);
+ evhttp_accept_socket (ctx->http, worker->cf->listen_sock);
+
+ if (ctx->use_ssl) {
+#ifdef HAVE_WEBUI_SSL
+ if (ctx->ssl_cert && ctx->ssl_key) {
+ webui_ssl_init (ctx);
+ }
+ else {
+ msg_err ("ssl cannot be enabled without key and cert for this server");
+ }
+#else
+ msg_err ("http ssl is not supported by this libevent version");
+#endif
+ }
+
+ /* Add callbacks for different methods */
+ evhttp_set_cb (ctx->http, PATH_AUTH, http_handle_auth, ctx);
+
+ ctx->resolver = dns_resolver_init (ctx->ev_base, worker->srv->cfg);
+
+ /* Maps events */
+ start_map_watch (worker->srv->cfg, ctx->ev_base);
+
+ event_base_loop (ctx->ev_base, 0);
+
+ close_log (rspamd_main->logger);
+ exit (EXIT_SUCCESS);
+}
projects / thirdparty / rspamd.git / commitdiff
