+++ /dev/null
-/*
- * Copyright (c) 2009-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 BY AUTHOR ''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 "map.h"
-#include "dns.h"
-#include "tokenizers/tokenizers.h"
-#include "classifiers/classifiers.h"
-#include "binlog.h"
-#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 (struct rspamd_config *cfg);
-void start_controller (struct rspamd_worker *worker);
-
-worker_t controller_worker = {
- "controller", /* Name */
- init_controller, /* Init function */
- start_controller, /* Start function */
- TRUE, /* Has socket */
- FALSE, /* Non unique */
- FALSE, /* Non threaded */
- TRUE, /* Killable */
- SOCK_STREAM /* TCP socket */
-};
-
-enum command_type {
- COMMAND_PASSWORD,
- COMMAND_QUIT,
- COMMAND_RELOAD,
- COMMAND_STAT,
- COMMAND_STAT_RESET,
- COMMAND_SHUTDOWN,
- COMMAND_UPTIME,
- COMMAND_LEARN,
- COMMAND_LEARN_SPAM,
- COMMAND_LEARN_HAM,
- COMMAND_HELP,
- COMMAND_COUNTERS,
- COMMAND_SYNC,
- COMMAND_WEIGHTS,
- COMMAND_GET,
- COMMAND_POST,
- COMMAND_ADD_SYMBOL,
- COMMAND_ADD_ACTION
-};
-
-struct controller_command {
- gchar *command;
- gboolean privilleged;
- enum command_type type;
-};
-
-struct custom_controller_command {
- const gchar *command;
- gboolean privilleged;
- gboolean require_message;
- controller_func_t handler;
-};
-
-struct rspamd_controller_ctx {
- char *password;
- 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[] = {
- {"password", FALSE, COMMAND_PASSWORD},
- {"quit", FALSE, COMMAND_QUIT},
- {"reload", TRUE, COMMAND_RELOAD},
- {"stat", FALSE, COMMAND_STAT},
- {"stat_reset", TRUE, COMMAND_STAT_RESET},
- {"shutdown", TRUE, COMMAND_SHUTDOWN},
- {"uptime", FALSE, COMMAND_UPTIME},
- {"learn", TRUE, COMMAND_LEARN},
- {"weights", FALSE, COMMAND_WEIGHTS},
- {"help", FALSE, COMMAND_HELP},
- {"counters", FALSE, COMMAND_COUNTERS},
- {"sync", FALSE, COMMAND_SYNC},
- {"learn_spam", TRUE, COMMAND_LEARN_SPAM},
- {"learn_ham", TRUE, COMMAND_LEARN_HAM},
- {"get", FALSE, COMMAND_GET},
- {"post", FALSE, COMMAND_POST},
- {"add_symbol", TRUE, COMMAND_ADD_SYMBOL},
- {"add_action", TRUE, COMMAND_ADD_ACTION}
-};
-
-static GList *custom_commands = NULL;
-
-static time_t start_time;
-
-static gchar greetingbuf[1024];
-
-static gboolean controller_write_socket (void *arg);
-
-static void
-free_session (void *ud)
-{
- GList *part;
- struct mime_part *p;
- struct controller_session *session = ud;
-
- msg_debug ("freeing session %p", session);
-
- while ((part = g_list_first (session->parts))) {
- session->parts = g_list_remove_link (session->parts, part);
- p = (struct mime_part *)part->data;
- g_byte_array_free (p->content, FALSE);
- g_list_free_1 (part);
- }
- rspamd_remove_dispatcher (session->dispatcher);
-
- if (session->kwargs) {
- g_hash_table_destroy (session->kwargs);
- }
-
- close (session->sock);
-
- rspamd_mempool_delete (session->session_pool);
- g_slice_free1 (sizeof (struct controller_session), session);
-}
-
-static gboolean
-restful_write_reply (gint error_code, const gchar *err_message,
- const gchar *buf, gsize buflen, rspamd_io_dispatcher_t *d)
-{
- static gchar hbuf[256];
- gint r;
-
- r = rspamd_snprintf (hbuf, sizeof (hbuf),
- "HTTP/1.0 %d %s" CRLF "Version: " RVERSION CRLF,
- error_code, err_message ? err_message : "OK");
- if (buflen > 0) {
- r += rspamd_snprintf (hbuf + r,
- sizeof (hbuf) - r,
- "Content-Length: %z" CRLF,
- buflen);
- }
- r += rspamd_snprintf (hbuf + r, sizeof (hbuf) - r, CRLF);
-
- if (buf != NULL) {
- if (!rspamd_dispatcher_write (d, hbuf, r, TRUE, TRUE)) {
- return FALSE;
- }
- return rspamd_dispatcher_write (d, buf, buflen, FALSE, FALSE);
- }
- else {
- if (!rspamd_dispatcher_write (d, hbuf, r, FALSE, TRUE)) {
- return FALSE;
- }
- }
-
- return TRUE;
-}
-
-static gboolean
-restful_write_reply_string (gint error_code, const gchar *err_message,
- GString *buf, rspamd_io_dispatcher_t *d)
-{
- static gchar hbuf[256];
- gint r;
-
- r = rspamd_snprintf (hbuf, sizeof (hbuf),
- "HTTP/1.0 %d %s" CRLF "Version: " RVERSION CRLF,
- error_code, err_message ? err_message : "OK");
- if (buf->len > 0) {
- r += rspamd_snprintf (hbuf + r,
- sizeof (hbuf) - r,
- "Content-Length: %z" CRLF,
- buf->len);
- }
- r += rspamd_snprintf (hbuf + r, sizeof (hbuf) - r, CRLF);
-
- if (buf != NULL) {
- if (!rspamd_dispatcher_write (d, hbuf, r, TRUE, TRUE)) {
- return FALSE;
- }
- return rspamd_dispatcher_write_string (d, buf, FALSE, TRUE);
- }
- else {
- if (!rspamd_dispatcher_write (d, hbuf, r, FALSE, TRUE)) {
- return FALSE;
- }
- }
-
- return TRUE;
-}
-
-static gint
-check_auth (struct controller_command *cmd, struct controller_session *session)
-{
- gchar out_buf[128];
- gint r;
-
- if (cmd->privilleged && !session->authorized) {
- if (!session->restful) {
- r = rspamd_snprintf (out_buf,
- sizeof (out_buf),
- "not authorized" CRLF);
- if (!rspamd_dispatcher_write (session->dispatcher, out_buf, r,
- FALSE, FALSE)) {
- return 0;
- }
- }
- else {
- (void)restful_write_reply (403,
- "Not authorized",
- NULL,
- 0,
- session->dispatcher);
- }
- return 0;
- }
-
- return 1;
-}
-
-static gboolean
-write_whole_statfile (struct controller_session *session,
- gchar *symbol,
- struct rspamd_classifier_config *ccf)
-{
- stat_file_t *statfile;
- struct rspamd_statfile_config *st;
- gchar out_buf[BUFSIZ];
- guint i;
- guint64 rev, ti, len, pos, blocks;
- gchar *out;
- struct rspamd_binlog_element log_elt;
- struct stat_file_block *stat_elt;
-
- statfile = get_statfile_by_symbol (session->worker->srv->statfile_pool, ccf,
- symbol, &st, FALSE);
- if (statfile == NULL) {
- return FALSE;
- }
-
- /* Begin to copy all blocks into array */
- statfile_get_revision (statfile, &rev, (time_t *)&ti);
- if (ti == 0) {
- /* Not tracked file */
- ti = time (NULL);
- statfile_set_revision (statfile, rev, ti);
- }
- msg_info ("send a whole statfile %s with version %uL to slave", symbol,
- rev);
-
- blocks = statfile_get_total_blocks (statfile);
- len = blocks * sizeof (struct rspamd_binlog_element);
- out = rspamd_mempool_alloc (session->session_pool, len);
-
- for (i = 0, pos = 0; i < blocks; i++) {
- stat_elt =
- (struct stat_file_block *)((u_char *)statfile->map +
- statfile->seek_pos + i * sizeof (struct stat_file_block));
- if (fabs (stat_elt->value) > 0.001) {
- /* Write only those values which value is not 0 */
- log_elt.h1 = stat_elt->hash1;
- log_elt.h2 = stat_elt->hash2;
- log_elt.value = stat_elt->value;
-
- memcpy (out + pos, &log_elt, sizeof (log_elt));
- pos += sizeof (struct rspamd_binlog_element);
- }
- }
-
- i = rspamd_snprintf (out_buf,
- sizeof (out_buf),
- "%uL %uL %uL" CRLF,
- rev,
- ti,
- pos);
- if (!rspamd_dispatcher_write (session->dispatcher, out_buf, i, TRUE,
- FALSE)) {
- return FALSE;
- }
-
- if (!rspamd_dispatcher_write (session->dispatcher, out, pos, TRUE, TRUE)) {
- return FALSE;
- }
-
- return TRUE;
-}
-
-static gboolean
-process_sync_command (struct controller_session *session, gchar **args)
-{
- gchar out_buf[BUFSIZ], *arg, *err_str, *symbol;
- gint r;
- guint64 rev, time;
- struct rspamd_statfile_config *st = NULL;
- struct rspamd_classifier_config *ccf;
- GList *cur;
- struct rspamd_binlog *binlog;
- GByteArray *data = NULL;
-
- arg = *args;
- if (!arg || *arg == '\0') {
- msg_info ("bad arguments to sync command, need symbol");
- return FALSE;
- }
- symbol = arg;
- arg = *(args + 1);
- if (!arg || *arg == '\0') {
- msg_info ("bad arguments to sync command, need revision");
- return FALSE;
- }
- rev = strtoull (arg, &err_str, 10);
- if (err_str && *err_str != 0) {
- msg_info ("bad arguments to sync command: %s", arg);
- return FALSE;
- }
- arg = *(args + 2);
- if (!arg || *arg == '\0') {
- msg_info ("bad arguments to sync command, need time");
- return FALSE;
- }
- time = strtoull (arg, &err_str, 10);
- if (err_str && *err_str != 0) {
- msg_info ("bad arguments to sync command: %s", arg);
- return FALSE;
- }
-
- ccf = g_hash_table_lookup (session->cfg->classifiers_symbols, symbol);
- if (ccf == NULL) {
- msg_info ("bad symbol: %s", symbol);
- return FALSE;
- }
-
- cur = g_list_first (ccf->statfiles);
- while (cur) {
- st = cur->data;
- if (strcmp (symbol, st->symbol) == 0) {
- break;
- }
- st = NULL;
- cur = g_list_next (cur);
- }
- if (st == NULL) {
- msg_info ("bad symbol: %s", symbol);
- return FALSE;
- }
-
- binlog = get_binlog_by_statfile (st);
- if (binlog == NULL) {
- msg_info ("cannot open binlog: %s", symbol);
- return FALSE;
- }
-
- while (binlog_sync (binlog, rev, &time, &data)) {
- rev++;
- r = rspamd_snprintf (out_buf,
- sizeof (out_buf),
- "%uL %uL %z" CRLF,
- rev,
- time,
- data->len);
- if (!rspamd_dispatcher_write (session->dispatcher, out_buf, r, FALSE,
- FALSE)) {
- if (data != NULL) {
- g_free (data);
- }
- return FALSE;
- }
- if (data->data != NULL) {
- if (!rspamd_dispatcher_write (session->dispatcher, data->data,
- data->len, TRUE, FALSE)) {
- if (data != NULL) {
- g_free (data);
- }
- return FALSE;
- }
- }
- }
-
- if (time == 0) {
- if (data != NULL) {
- g_free (data);
- }
- return write_whole_statfile (session, symbol, ccf);
- }
-
- if (data != NULL) {
- g_free (data);
- }
-
- return TRUE;
-}
-
-static gboolean
-process_counters_command (struct controller_session *session)
-{
- GList *cur;
- struct cache_item *item;
- struct symbols_cache *cache;
- GString *out;
-
- cache = session->cfg->cache;
- out = g_string_sized_new (BUFSIZ);
-
- if (!session->restful) {
- rspamd_printf_gstring (out, "Rspamd counters:" CRLF);
- }
-
- if (cache != NULL) {
- cur = cache->negative_items;
- while (cur) {
- item = cur->data;
- if (!item->is_callback) {
- rspamd_printf_gstring (out, "%s %.2f %d %.3f" CRLF,
- item->s->symbol, item->s->weight,
- item->s->frequency, item->s->avg_time);
- }
- cur = g_list_next (cur);
- }
- cur = cache->static_items;
- while (cur) {
- item = cur->data;
- if (!item->is_callback) {
- rspamd_printf_gstring (out, "%s %.2f %d %.3f" CRLF,
- item->s->symbol, item->s->weight,
- item->s->frequency, item->s->avg_time);
- }
- cur = g_list_next (cur);
- }
- }
-
- if (!session->restful) {
- return rspamd_dispatcher_write_string (session->dispatcher,
- out,
- FALSE,
- TRUE);
- }
- else {
- return restful_write_reply_string (200, NULL, out, session->dispatcher);
- }
-}
-
-static gboolean
-process_stat_command (struct controller_session *session, gboolean do_reset)
-{
- GString *out;
- gint i;
- guint64 used, total, rev, ham = 0, spam = 0;
- time_t ti;
- rspamd_mempool_stat_t mem_st;
- struct rspamd_classifier_config *ccf;
- stat_file_t *statfile;
- struct rspamd_statfile_config *st;
- GList *cur_cl, *cur_st;
- struct rspamd_stat *stat, stat_copy;
-
- rspamd_mempool_stat (&mem_st);
- memcpy (&stat_copy, session->worker->srv->stat, sizeof (stat_copy));
- stat = &stat_copy;
- out = g_string_sized_new (BUFSIZ);
- rspamd_printf_gstring (out,
- "Messages scanned: %ud" CRLF,
- stat->messages_scanned);
- if (stat->messages_scanned > 0) {
- for (i = METRIC_ACTION_REJECT; i <= METRIC_ACTION_NOACTION; i++) {
- rspamd_printf_gstring (out,
- "Messages with action %s: %ud, %.2f%%" CRLF,
- rspamd_action_to_str (
- i),
- stat->actions_stat[i],
- (double)stat->actions_stat[i] / (double)stat->messages_scanned *
- 100.);
- if (i < METRIC_ACTION_GREYLIST) {
- spam += stat->actions_stat[i];
- }
- else {
- ham += stat->actions_stat[i];
- }
- if (do_reset) {
- session->worker->srv->stat->actions_stat[i] = 0;
- }
- }
- rspamd_printf_gstring (out,
- "Messages treated as spam: %ud, %.2f%%" CRLF,
- spam,
- (double)spam / (double)stat->messages_scanned * 100.);
- rspamd_printf_gstring (out,
- "Messages treated as ham: %ud, %.2f%%" CRLF,
- ham,
- (double)ham / (double)stat->messages_scanned * 100.);
- }
- rspamd_printf_gstring (out,
- "Messages learned: %ud" CRLF,
- stat->messages_learned);
- rspamd_printf_gstring (out,
- "Connections count: %ud" CRLF,
- stat->connections_count);
- rspamd_printf_gstring (out,
- "Control connections count: %ud" CRLF,
- stat->control_connections_count);
- rspamd_printf_gstring (out,
- "Pools allocated: %z" CRLF,
- mem_st.pools_allocated);
- rspamd_printf_gstring (out,
- "Pools freed: %z" CRLF,
- mem_st.pools_freed);
- rspamd_printf_gstring (out,
- "Bytes allocated: %z" CRLF,
- mem_st.bytes_allocated);
- rspamd_printf_gstring (out,
- "Memory chunks allocated: %z" CRLF,
- mem_st.chunks_allocated);
- rspamd_printf_gstring (out,
- "Shared chunks allocated: %z" CRLF,
- mem_st.shared_chunks_allocated);
- rspamd_printf_gstring (out,
- "Chunks freed: %z" CRLF,
- mem_st.chunks_freed);
- rspamd_printf_gstring (out,
- "Oversized chunks: %z" CRLF,
- mem_st.oversized_chunks);
- rspamd_printf_gstring (out,
- "Fuzzy hashes stored: %ud" CRLF,
- stat->fuzzy_hashes);
- rspamd_printf_gstring (out,
- "Fuzzy hashes expired: %ud" CRLF,
- stat->fuzzy_hashes_expired);
- /* Now write statistics for each statfile */
- cur_cl = g_list_first (session->cfg->classifiers);
- while (cur_cl) {
- ccf = cur_cl->data;
- cur_st = g_list_first (ccf->statfiles);
- while (cur_st) {
- st = cur_st->data;
- if ((statfile =
- statfile_pool_is_open (session->worker->srv->statfile_pool,
- st->path)) == NULL) {
- statfile = statfile_pool_open (
- session->worker->srv->statfile_pool,
- st->path,
- st->size,
- FALSE);
- }
- if (statfile) {
- used = statfile_get_used_blocks (statfile);
- total = statfile_get_total_blocks (statfile);
- statfile_get_revision (statfile, &rev, &ti);
- if (total != (guint64) - 1 && used != (guint64) - 1) {
- if (st->label) {
- rspamd_printf_gstring (out,
- "Statfile: %s <%s> (version %uL); length: %Hz; free blocks: %uL; total blocks: %uL; free: %.2f%%" CRLF,
- st->symbol,
- st->label,
- rev,
- st->size,
- (total - used),
- total,
- (double)((double)(total -
- used) / (double)total) * 100.);
- }
- else {
- rspamd_printf_gstring (out,
- "Statfile: %s (version %uL); length: %Hz; free blocks: %uL; total blocks: %uL; free: %.2f%%" CRLF,
- st->symbol,
- rev,
- st->size,
- (total - used),
- total,
- (double)((double)(total -
- used) / (double)total) * 100.);
- }
- }
- }
- cur_st = g_list_next (cur_st);
- }
- cur_cl = g_list_next (cur_cl);
- }
-
- if (do_reset) {
- session->worker->srv->stat->messages_scanned = 0;
- session->worker->srv->stat->messages_learned = 0;
- session->worker->srv->stat->connections_count = 0;
- session->worker->srv->stat->control_connections_count = 0;
- }
-
- if (!session->restful) {
- return rspamd_dispatcher_write_string (session->dispatcher,
- out,
- FALSE,
- TRUE);
- }
- else {
- return restful_write_reply_string (200, NULL, out, session->dispatcher);
- }
-}
-
-static gboolean
-process_dynamic_conf_command (gchar **cmd_args,
- struct controller_session *session,
- gboolean is_action)
-{
- struct rspamd_config *cfg = session->cfg;
- gchar *arg, *metric, *name, *err_str;
- gdouble value;
- gboolean res;
- guint real_act;
-
- if (cfg->dynamic_conf == NULL) {
- if (!session->restful) {
- return rspamd_dispatcher_write (session->dispatcher,
- "dynamic config is not specified" CRLF,
- 0,
- FALSE,
- TRUE);
- }
- else {
- return restful_write_reply (500,
- "dynamic config is not specified",
- NULL,
- 0,
- session->dispatcher);
- }
- }
-
- if (session->restful) {
- if ((arg = g_hash_table_lookup (session->kwargs, "metric")) == NULL) {
- metric = DEFAULT_METRIC;
- }
- else {
- metric = arg;
- }
- if ((arg = g_hash_table_lookup (session->kwargs, "name")) == NULL) {
- goto invalid_arguments;
- }
- name = arg;
- if ((arg = g_hash_table_lookup (session->kwargs, "value")) == NULL) {
- goto invalid_arguments;
- }
- value = strtod (arg, &err_str);
- if (err_str && *err_str != '\0') {
- msg_info ("double value is invalid: %s", arg);
- goto invalid_arguments;
- }
- }
- else {
- if (cmd_args[0] == NULL || cmd_args[1] == NULL) {
- goto invalid_arguments;
- }
- if (cmd_args[2] == NULL) {
- metric = DEFAULT_METRIC;
- name = cmd_args[0];
- arg = cmd_args[1];
- value = strtod (arg, &err_str);
- if (err_str && *err_str != '\0') {
- msg_info ("double value is invalid: %s", arg);
- goto invalid_arguments;
- }
- }
- else {
- metric = cmd_args[0];
- name = cmd_args[1];
- arg = cmd_args[2];
- value = strtod (arg, &err_str);
- if (err_str && *err_str != '\0') {
- msg_info ("double value is invalid: %s", arg);
- goto invalid_arguments;
- }
- }
- }
-
- if (is_action) {
- if (!rspamd_action_from_str (name, &real_act)) {
- msg_info ("invalid action string: %s", name);
- res = FALSE;
- }
- else {
- res = add_dynamic_action (cfg, metric, real_act, value);
- }
- }
- else {
- res = add_dynamic_symbol (cfg, metric, name, value);
- }
-
- if (res) {
-
- res = dump_dynamic_config (cfg);
- if (res) {
- if (!session->restful) {
- return rspamd_dispatcher_write (session->dispatcher,
- "OK" CRLF,
- 0,
- FALSE,
- TRUE);
- }
- else {
- return restful_write_reply (200,
- "OK",
- NULL,
- 0,
- session->dispatcher);
- }
- }
- else {
- if (!session->restful) {
- return rspamd_dispatcher_write (session->dispatcher,
- "Error dumping dynamic config" CRLF,
- 0,
- FALSE,
- TRUE);
- }
- else {
- return restful_write_reply (500,
- "Error dumping dynamic config",
- NULL,
- 0,
- session->dispatcher);
- }
- }
- }
- else {
- if (!session->restful) {
- return rspamd_dispatcher_write (session->dispatcher,
- "Cannot add dynamic rule" CRLF,
- 0,
- FALSE,
- TRUE);
- }
- else {
- return restful_write_reply (500,
- "Cannot add dynamic rule",
- NULL,
- 0,
- session->dispatcher);
- }
- }
-
-invalid_arguments:
- if (!session->restful) {
- return rspamd_dispatcher_write (session->dispatcher,
- "Invalid arguments" CRLF,
- 0,
- FALSE,
- TRUE);
- }
- else {
- return restful_write_reply (500,
- "Invalid arguments",
- NULL,
- 0,
- session->dispatcher);
- }
-}
-
-static gboolean
-process_command (struct controller_command *cmd,
- gchar **cmd_args,
- struct controller_session *session)
-{
- gchar out_buf[BUFSIZ], *arg, *err_str;
- gint r = 0, days, hours, minutes;
- time_t uptime;
- guint32 size = 0;
- struct rspamd_classifier_config *cl;
- struct rspamd_controller_ctx *ctx = session->worker->ctx;
-
- switch (cmd->type) {
- case COMMAND_GET:
- case COMMAND_POST:
- session->restful = TRUE;
- session->state = STATE_HEADER;
- session->kwargs = g_hash_table_new (rspamd_strcase_hash,
- rspamd_strcase_equal);
- break;
- case COMMAND_PASSWORD:
- arg = *cmd_args;
- if (!arg || *arg == '\0') {
- msg_debug ("empty password passed");
- r = rspamd_snprintf (out_buf,
- sizeof (out_buf),
- "password command requires one argument" CRLF);
- if (!rspamd_dispatcher_write (session->dispatcher, out_buf, r,
- FALSE, FALSE)) {
- return FALSE;
- }
- return TRUE;
- }
- if (ctx->password == NULL) {
- r = rspamd_snprintf (out_buf,
- sizeof (out_buf),
- "password command disabled in config, authorized access granted" CRLF);
- if (!rspamd_dispatcher_write (session->dispatcher, out_buf, r,
- FALSE, FALSE)) {
- return FALSE;
- }
- return TRUE;
- }
- if (strncmp (arg, ctx->password, strlen (arg)) == 0) {
- session->authorized = 1;
- r = rspamd_snprintf (out_buf,
- sizeof (out_buf),
- "password accepted" CRLF);
- if (!rspamd_dispatcher_write (session->dispatcher, out_buf, r,
- FALSE, FALSE)) {
- return FALSE;
- }
- }
- else {
- session->authorized = 0;
- r = rspamd_snprintf (out_buf,
- sizeof (out_buf),
- "password NOT accepted" CRLF);
- if (!rspamd_dispatcher_write (session->dispatcher, out_buf, r,
- FALSE, FALSE)) {
- return FALSE;
- }
- }
- break;
- case COMMAND_QUIT:
- session->state = STATE_QUIT;
- break;
- case COMMAND_RELOAD:
- if (check_auth (cmd, session)) {
- r = rspamd_snprintf (out_buf,
- sizeof (out_buf),
- "reload request sent" CRLF);
- if (!session->restful) {
- if (!rspamd_dispatcher_write (session->dispatcher, out_buf, r,
- FALSE, FALSE)) {
- return FALSE;
- }
- }
- else {
- if (!restful_write_reply (200, out_buf, NULL, 0,
- session->dispatcher)) {
- return FALSE;
- }
- }
- kill (getppid (), SIGHUP);
- }
- break;
- case COMMAND_STAT:
- if (check_auth (cmd, session)) {
- return process_stat_command (session, FALSE);
- }
- break;
- case COMMAND_STAT_RESET:
- if (check_auth (cmd, session)) {
- return process_stat_command (session, TRUE);
- }
- break;
- case COMMAND_SHUTDOWN:
- if (check_auth (cmd, session)) {
- r = rspamd_snprintf (out_buf,
- sizeof (out_buf),
- "shutdown request sent" CRLF);
- if (!session->restful) {
- if (!rspamd_dispatcher_write (session->dispatcher, out_buf, r,
- FALSE, FALSE)) {
- return FALSE;
- }
- }
- else {
- if (!restful_write_reply (200, out_buf, NULL, 0,
- session->dispatcher)) {
- return FALSE;
- }
- }
- kill (getppid (), SIGTERM);
- }
- break;
- case COMMAND_UPTIME:
- if (check_auth (cmd, session)) {
- uptime = time (NULL) - 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;
- r = rspamd_snprintf (out_buf,
- sizeof (out_buf),
- "%d day%s %d hour%s %d minute%s" CRLF,
- 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) {
- r = rspamd_snprintf (out_buf,
- sizeof (out_buf),
- "%d second%s" CRLF,
- (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;
- r = rspamd_snprintf (out_buf,
- sizeof (out_buf),
- "%d hour%s %d minute%s %d second%s" CRLF,
- hours,
- hours > 1 ? "s" : " ",
- minutes,
- minutes > 1 ? "s" : " ",
- (gint)uptime,
- uptime > 1 ? "s" : " ");
- }
- if (!session->restful) {
- if (!rspamd_dispatcher_write (session->dispatcher, out_buf, r,
- FALSE, FALSE)) {
- return FALSE;
- }
- }
- else {
- if (!restful_write_reply (200, NULL, out_buf, r,
- session->dispatcher)) {
- return FALSE;
- }
- }
- }
- break;
- case COMMAND_LEARN_SPAM:
- if (check_auth (cmd, session)) {
- if (!session->restful) {
- arg = *cmd_args;
- if (!arg || *arg == '\0') {
- msg_debug ("no statfile specified in learn command");
- r = rspamd_snprintf (out_buf,
- sizeof (out_buf),
- "learn_spam command requires at least two arguments: classifier name and a message's size" CRLF);
- if (!rspamd_dispatcher_write (session->dispatcher, out_buf,
- r, FALSE, FALSE)) {
- return FALSE;
- }
- return TRUE;
- }
- arg = *(cmd_args + 1);
- if (arg == NULL || *arg == '\0') {
- msg_debug ("no message size specified in learn command");
- r = rspamd_snprintf (out_buf,
- sizeof (out_buf),
- "learn_spam command requires at least two arguments: classifier name and a message's size" CRLF);
- if (!rspamd_dispatcher_write (session->dispatcher, out_buf,
- r, FALSE, FALSE)) {
- return FALSE;
- }
- return TRUE;
- }
- size = strtoul (arg, &err_str, 10);
- if (err_str && *err_str != '\0') {
- msg_debug ("message size is invalid: %s", arg);
- r = rspamd_snprintf (out_buf,
- sizeof (out_buf),
- "learn size is invalid" CRLF);
- if (!rspamd_dispatcher_write (session->dispatcher, out_buf,
- r, FALSE, FALSE)) {
- return FALSE;
- }
- return TRUE;
- }
- cl = rspamd_config_find_classifier (session->cfg, *cmd_args);
- }
- else {
- if ((arg =
- g_hash_table_lookup (session->kwargs,
- "classifier")) == NULL) {
- msg_debug ("no classifier specified in learn command");
- r = rspamd_snprintf (out_buf,
- sizeof (out_buf),
- "learn_spam command requires at least two arguments: classifier name and a message's size" CRLF);
- if (!restful_write_reply (500, out_buf, NULL, 0,
- session->dispatcher)) {
- return FALSE;
- }
- return TRUE;
- }
- else {
- cl = rspamd_config_find_classifier (session->cfg, arg);
- }
- if ((arg =
- g_hash_table_lookup (session->kwargs,
- "content-length")) == NULL) {
- msg_debug ("no size specified in learn command");
- r = rspamd_snprintf (out_buf,
- sizeof (out_buf),
- "learn_spam command requires at least two arguments: classifier name and a message's size" CRLF);
- if (!restful_write_reply (500, out_buf, NULL, 0,
- session->dispatcher)) {
- return FALSE;
- }
- }
- else {
- size = strtoul (arg, &err_str, 10);
- if (err_str && *err_str != '\0') {
- msg_debug ("message size is invalid: %s", arg);
- r = rspamd_snprintf (out_buf,
- sizeof (out_buf),
- "learn size is invalid" CRLF);
- if (!restful_write_reply (500, out_buf, NULL, 0,
- session->dispatcher)) {
- return FALSE;
- }
- return TRUE;
- }
- }
- }
-
- session->learn_classifier = cl;
-
- /* By default learn positive */
- session->in_class = TRUE;
- rspamd_set_dispatcher_policy (session->dispatcher,
- BUFFER_CHARACTER,
- size);
- session->state = STATE_LEARN_SPAM_PRE;
- }
- break;
- case COMMAND_LEARN_HAM:
- if (check_auth (cmd, session)) {
- if (!session->restful) {
- arg = *cmd_args;
- if (!arg || *arg == '\0') {
- msg_debug ("no statfile specified in learn command");
- r = rspamd_snprintf (out_buf,
- sizeof (out_buf),
- "learn_spam command requires at least two arguments: classifier name and a message's size" CRLF);
- if (!rspamd_dispatcher_write (session->dispatcher, out_buf,
- r, FALSE, FALSE)) {
- return FALSE;
- }
- return TRUE;
- }
- arg = *(cmd_args + 1);
- if (arg == NULL || *arg == '\0') {
- msg_debug ("no message size specified in learn command");
- r = rspamd_snprintf (out_buf,
- sizeof (out_buf),
- "learn_spam command requires at least two arguments: classifier name and a message's size" CRLF);
- if (!rspamd_dispatcher_write (session->dispatcher, out_buf,
- r, FALSE, FALSE)) {
- return FALSE;
- }
- return TRUE;
- }
- size = strtoul (arg, &err_str, 10);
- if (err_str && *err_str != '\0') {
- msg_debug ("message size is invalid: %s", arg);
- r = rspamd_snprintf (out_buf,
- sizeof (out_buf),
- "learn size is invalid" CRLF);
- if (!rspamd_dispatcher_write (session->dispatcher, out_buf,
- r, FALSE, FALSE)) {
- return FALSE;
- }
- return TRUE;
- }
- cl = rspamd_config_find_classifier (session->cfg, *cmd_args);
- }
- else {
- if ((arg =
- g_hash_table_lookup (session->kwargs,
- "classifier")) == NULL) {
- msg_debug ("no classifier specified in learn command");
- r = rspamd_snprintf (out_buf,
- sizeof (out_buf),
- "learn_spam command requires at least two arguments: classifier name and a message's size" CRLF);
- if (!restful_write_reply (500, out_buf, NULL, 0,
- session->dispatcher)) {
- return FALSE;
- }
- return TRUE;
- }
- else {
- cl = rspamd_config_find_classifier (session->cfg, arg);
- }
- if ((arg =
- g_hash_table_lookup (session->kwargs,
- "content-length")) == NULL) {
- msg_debug ("no size specified in learn command");
- r = rspamd_snprintf (out_buf,
- sizeof (out_buf),
- "learn_spam command requires at least two arguments: classifier name and a message's size" CRLF);
- if (!restful_write_reply (500, out_buf, NULL, 0,
- session->dispatcher)) {
- return FALSE;
- }
- }
- else {
- size = strtoul (arg, &err_str, 10);
- if (err_str && *err_str != '\0') {
- msg_debug ("message size is invalid: %s", arg);
- r = rspamd_snprintf (out_buf,
- sizeof (out_buf),
- "learn size is invalid" CRLF);
- if (!restful_write_reply (500, out_buf, NULL, 0,
- session->dispatcher)) {
- return FALSE;
- }
- return TRUE;
- }
- }
- }
-
- session->learn_classifier = cl;
-
- /* By default learn positive */
- session->in_class = FALSE;
- rspamd_set_dispatcher_policy (session->dispatcher,
- BUFFER_CHARACTER,
- size);
- session->state = STATE_LEARN_SPAM_PRE;
- }
- break;
- case COMMAND_LEARN:
- if (check_auth (cmd, session)) {
- /* TODO: remove this command as currenly it should not be used anywhere */
- arg = *cmd_args;
- if (!arg || *arg == '\0') {
- msg_debug ("no statfile specified in learn command");
- r = rspamd_snprintf (out_buf,
- sizeof (out_buf),
- "learn command requires at least two arguments: stat filename and its size" CRLF);
- if (!rspamd_dispatcher_write (session->dispatcher, out_buf, r,
- FALSE, FALSE)) {
- return FALSE;
- }
- return TRUE;
- }
- arg = *(cmd_args + 1);
- if (arg == NULL || *arg == '\0') {
- msg_debug ("no message size specified in learn command");
- r = rspamd_snprintf (out_buf,
- sizeof (out_buf),
- "learn command requires at least two arguments: symbol and message size" CRLF);
- if (!rspamd_dispatcher_write (session->dispatcher, out_buf, r,
- FALSE, FALSE)) {
- return FALSE;
- }
- return TRUE;
- }
- size = strtoul (arg, &err_str, 10);
- if (err_str && *err_str != '\0') {
- msg_debug ("message size is invalid: %s", arg);
- r = rspamd_snprintf (out_buf,
- sizeof (out_buf),
- "learn size is invalid" CRLF);
- if (!rspamd_dispatcher_write (session->dispatcher, out_buf, r,
- FALSE, FALSE)) {
- return FALSE;
- }
- return TRUE;
- }
-
- session->learn_symbol = rspamd_mempool_strdup (
- session->session_pool,
- *cmd_args);
- cl = g_hash_table_lookup (session->cfg->classifiers_symbols,
- *cmd_args);
-
- session->learn_classifier = cl;
-
- /* By default learn positive */
- session->in_class = 1;
- session->learn_multiplier = 1.;
- /* Get all arguments */
- while (*cmd_args++) {
- arg = *cmd_args;
- if (arg && *arg == '-') {
- switch (*(arg + 1)) {
- case 'r':
- arg = *(cmd_args + 1);
- if (!arg || *arg == '\0') {
- r = rspamd_snprintf (out_buf,
- sizeof (out_buf),
- "recipient is not defined" CRLF);
- if (!rspamd_dispatcher_write (session->dispatcher,
- out_buf, r, FALSE, FALSE)) {
- return FALSE;
- }
- }
- session->learn_rcpt = rspamd_mempool_strdup (
- session->session_pool,
- arg);
- break;
- case 'f':
- arg = *(cmd_args + 1);
- if (!arg || *arg == '\0') {
- r = rspamd_snprintf (out_buf,
- sizeof (out_buf),
- "from is not defined" CRLF);
- if (!rspamd_dispatcher_write (session->dispatcher,
- out_buf, r, FALSE, FALSE)) {
- return FALSE;
- }
- }
- session->learn_from = rspamd_mempool_strdup (
- session->session_pool,
- arg);
- break;
- case 'n':
- session->in_class = 0;
- break;
- case 'm':
- arg = *(cmd_args + 1);
- if (!arg || *arg == '\0') {
- r = rspamd_snprintf (out_buf,
- sizeof (out_buf),
- "multiplier is not defined" CRLF);
- if (!rspamd_dispatcher_write (session->dispatcher,
- out_buf, r, FALSE, FALSE)) {
- return FALSE;
- }
- }
- else {
- session->learn_multiplier = strtod (arg, NULL);
- }
- break;
- default:
- r = rspamd_snprintf (out_buf,
- sizeof (out_buf),
- "tokenizer is not defined" CRLF);
- if (!rspamd_dispatcher_write (session->dispatcher,
- out_buf, r, FALSE, FALSE)) {
- return FALSE;
- }
- return TRUE;
- }
- }
- }
- rspamd_set_dispatcher_policy (session->dispatcher,
- BUFFER_CHARACTER,
- size);
- session->state = STATE_LEARN;
- }
- break;
-
- case COMMAND_WEIGHTS:
- /* TODO: remove this command as currenly it should not be used anywhere */
- arg = *cmd_args;
- if (!arg || *arg == '\0') {
- msg_debug ("no statfile specified in weights command");
- r = rspamd_snprintf (out_buf,
- sizeof (out_buf),
- "weights command requires two arguments: statfile and message size" CRLF);
- if (!rspamd_dispatcher_write (session->dispatcher, out_buf, r,
- FALSE, FALSE)) {
- return FALSE;
- }
- return TRUE;
- }
- arg = *(cmd_args + 1);
- if (arg == NULL || *arg == '\0') {
- msg_debug ("no message size specified in weights command");
- r = rspamd_snprintf (out_buf,
- sizeof (out_buf),
- "weights command requires two arguments: statfile and message size" CRLF);
- if (!rspamd_dispatcher_write (session->dispatcher, out_buf, r,
- FALSE, FALSE)) {
- return FALSE;
- }
- return TRUE;
- }
- size = strtoul (arg, &err_str, 10);
- if (err_str && *err_str != '\0') {
- msg_debug ("message size is invalid: %s", arg);
- r = rspamd_snprintf (out_buf,
- sizeof (out_buf),
- "message size is invalid" CRLF);
- if (!rspamd_dispatcher_write (session->dispatcher, out_buf, r,
- FALSE, FALSE)) {
- return FALSE;
- }
- return TRUE;
- }
-
- cl = g_hash_table_lookup (session->cfg->classifiers_symbols, *cmd_args);
- if (cl == NULL) {
- r = rspamd_snprintf (out_buf,
- sizeof (out_buf),
- "statfile %s is not defined" CRLF,
- *cmd_args);
- if (!rspamd_dispatcher_write (session->dispatcher, out_buf, r,
- FALSE, FALSE)) {
- return FALSE;
- }
- return TRUE;
-
- }
- session->learn_classifier = cl;
-
- rspamd_set_dispatcher_policy (session->dispatcher,
- BUFFER_CHARACTER,
- size);
- session->state = STATE_WEIGHTS;
- break;
- case COMMAND_SYNC:
- if (!process_sync_command (session, cmd_args)) {
- r = rspamd_snprintf (out_buf, sizeof (out_buf), "FAIL" CRLF);
- if (!rspamd_dispatcher_write (session->dispatcher, out_buf, r,
- FALSE, FALSE)) {
- return FALSE;
- }
- return TRUE;
- }
- break;
- case COMMAND_HELP:
- r = rspamd_snprintf (out_buf,
- sizeof (out_buf),
- "Rspamd CLI commands (* - privileged command):" CRLF
- " help - this help message" CRLF
- "(*) learn <statfile> <size> [-r recipient] [-m multiplier] [-f from] [-n] - learn message to specified statfile" CRLF
- " quit - quit CLI session" CRLF
- " password <password> - authenticate yourself for privileged commands" CRLF
- "(*) reload - reload rspamd" CRLF
- "(*) shutdown - shutdown rspamd" CRLF
- " stat - show different rspamd stat" CRLF
- " sync - run synchronization of statfiles" CRLF
- " counters - show rspamd counters" CRLF
- " uptime - rspamd uptime" CRLF
- " weights <statfile> <size> - weight of message in all statfiles" CRLF);
- if (!session->restful) {
- if (!rspamd_dispatcher_write (session->dispatcher, out_buf, r,
- FALSE, FALSE)) {
- return FALSE;
- }
- }
- else {
- if (!restful_write_reply (200, NULL, out_buf, r,
- session->dispatcher)) {
- return FALSE;
- }
- }
- break;
- case COMMAND_COUNTERS:
- process_counters_command (session);
- break;
- case COMMAND_ADD_ACTION:
- if (check_auth (cmd, session)) {
- return process_dynamic_conf_command (cmd_args, session, TRUE);
- }
- break;
- case COMMAND_ADD_SYMBOL:
- if (check_auth (cmd, session)) {
- return process_dynamic_conf_command (cmd_args, session, FALSE);
- }
- break;
- }
- return TRUE;
-}
-
-static controller_func_t
-parse_custom_command (gchar *line,
- gchar **cmd_args,
- struct controller_session *session,
- gsize len)
-{
- GList *cur;
- struct custom_controller_command *cmd;
-
- if (len == 0) {
- len = strlen (line);
- }
- cur = custom_commands;
- while (cur) {
- cmd = cur->data;
- if (g_ascii_strncasecmp (cmd->command, line, len) == 0) {
- return cmd->handler;
- }
- cur = g_list_next (cur);
- }
-
- return NULL;
-}
-
-static struct controller_command *
-parse_normal_command (const gchar *line, gsize len)
-{
- guint i;
- struct controller_command *c;
-
- if (len == 0) {
- len = strlen (line);
- }
- for (i = 0; i < G_N_ELEMENTS (commands); i++) {
- c = &commands[i];
- if (g_ascii_strncasecmp (line, c->command, len) == 0) {
- return c;
- }
- }
-
- return NULL;
-}
-
-static gboolean
-process_header (f_str_t *line, struct controller_session *session)
-{
- gchar *headern;
- struct controller_command *command;
- struct rspamd_controller_ctx *ctx = session->worker->ctx;
- controller_func_t custom_handler;
-
- /* XXX: temporary workaround */
- headern = NULL;
-
- if (line == NULL || headern == NULL) {
- msg_warn ("bad header: %V", line);
- return FALSE;
- }
- /* Eat whitespaces */
- g_strstrip (headern);
- fstrstrip (line);
-
- if (*headern == 'c' || *headern == 'C') {
- if (g_ascii_strcasecmp (headern, "command") == 0) {
- /* This header is actually command */
- command = parse_normal_command (line->begin, line->len);
- if (command == NULL) {
- if ((custom_handler =
- parse_custom_command (line->begin, NULL, session,
- line->len)) == NULL) {
- msg_info ("bad command header: %V", line);
- return FALSE;
- }
- else {
- session->custom_handler = custom_handler;
- }
- }
- session->cmd = command;
- return TRUE;
- }
- }
- else if (*headern == 'p' || *headern == 'P') {
- /* Password header */
- if (g_ascii_strcasecmp (headern, "password") == 0) {
- if (ctx->password == NULL ||
- (line->len == strlen (ctx->password)
- && memcmp (line->begin, ctx->password, line->len) == 0)) {
- session->authorized = TRUE;
- }
- else {
- msg_info ("wrong password in controller command");
- }
- return TRUE;
- }
- }
-
- g_hash_table_insert (session->kwargs, headern,
- fstrcstr (line, session->session_pool));
-
- return TRUE;
-}
-
-/*
- * Called if all filters are processed, non-threaded and simple version
- */
-static gboolean
-fin_learn_task (void *arg)
-{
- struct rspamd_task *task = (struct rspamd_task *) arg;
-
- /* XXX: needs to be reworked */
-
- return TRUE;
-}
-
-/*
- * Called if session was restored inside fin callback
- */
-static void
-restore_learn_task (void *arg)
-{
- struct rspamd_task *task = (struct rspamd_task *) arg;
-
- /* Special state */
-}
-
-static gboolean
-controller_read_socket (f_str_t * in, void *arg)
-{
- struct controller_session *session = (struct controller_session *)arg;
- struct classifier_ctx *cls_ctx;
- gint len, i, r;
- gchar *s, **params, *cmd, out_buf[128];
- struct controller_command *command;
- struct rspamd_task *task;
- struct mime_text_part *part;
- GList *cur = NULL;
- GTree *tokens = NULL;
- GError *err = NULL;
- f_str_t c;
- controller_func_t custom_handler;
-
- switch (session->state) {
- case STATE_COMMAND:
- s = fstrcstr (in, session->session_pool);
- params = g_strsplit_set (s, " ", -1);
-
- rspamd_mempool_add_destructor (session->session_pool,
- (rspamd_mempool_destruct_t) g_strfreev, params);
-
- len = g_strv_length (params);
- if (len > 0) {
- cmd = g_strstrip (params[0]);
-
- command = parse_normal_command (cmd, 0);
- if (command != NULL) {
- if (!process_command (command, ¶ms[1], session)) {
- return FALSE;
- }
- }
- else {
- if ((custom_handler =
- parse_custom_command (cmd, ¶ms[1], session,
- 0)) == NULL) {
- msg_debug ("'%s'", cmd);
- i = rspamd_snprintf (out_buf,
- sizeof (out_buf),
- "Unknown command" CRLF);
- if (!rspamd_dispatcher_write (session->dispatcher, out_buf,
- i, FALSE, FALSE)) {
- return FALSE;
- }
- }
- else {
- custom_handler (¶ms[1], session);
- }
- }
- }
- if (session->state != STATE_LEARN && session->state !=
- STATE_LEARN_SPAM_PRE
- && session->state != STATE_WEIGHTS && session->state !=
- STATE_OTHER && session->state != STATE_HEADER) {
- if (!rspamd_dispatcher_write (session->dispatcher, END,
- sizeof (END) - 1, FALSE, TRUE)) {
- return FALSE;
- }
- if (session->state == STATE_QUIT) {
- destroy_session (session->s);
- return FALSE;
- }
- }
-
- break;
- case STATE_HEADER:
- if (in->len == 0) {
- /* End of headers */
- if (session->cmd == NULL && session->custom_handler == NULL) {
- i = rspamd_snprintf (out_buf,
- sizeof (out_buf),
- "HTTP/1.0 500 Bad command" CRLF CRLF);
- if (!rspamd_dispatcher_write (session->dispatcher, out_buf, i,
- FALSE, FALSE)) {
- return FALSE;
- }
- destroy_session (session->s);
- return FALSE;
- }
- /* Perform command */
- else if (session->cmd != NULL) {
- if (!process_command (session->cmd, NULL, session)) {
- msg_debug ("process command failed");
- return FALSE;
- }
- }
- else {
- session->custom_handler (NULL, session);
- }
- if (session->state != STATE_LEARN && session->state !=
- STATE_LEARN_SPAM_PRE
- && session->state != STATE_WEIGHTS && session->state !=
- STATE_OTHER) {
- msg_debug ("closing restful connection");
- destroy_session (session->s);
- return FALSE;
- }
- }
- else if (!process_header (in, session)) {
- i = rspamd_snprintf (out_buf,
- sizeof (out_buf),
- "HTTP/1.0 500 Bad header" CRLF CRLF);
- if (!rspamd_dispatcher_write (session->dispatcher, out_buf, i,
- FALSE, FALSE)) {
- return FALSE;
- }
- destroy_session (session->s);
- return FALSE;
- }
- break;
- case STATE_LEARN:
- session->learn_buf = in;
- task = rspamd_task_new (session->worker);
-
- task->msg = g_string_new_len (in->begin, in->len);
- task->ev_base = session->ev_base;
-
- r = process_message (task);
- if (r == -1) {
- msg_warn ("processing of message failed");
- rspamd_task_free (task, FALSE);
- session->state = STATE_REPLY;
- if (session->restful) {
- r = rspamd_snprintf (out_buf,
- sizeof (out_buf),
- "HTTP/1.0 500 Cannot process message" CRLF CRLF);
- }
- else {
- r = rspamd_snprintf (out_buf,
- sizeof (out_buf),
- "cannot process message" CRLF);
- }
- if (!rspamd_dispatcher_write (session->dispatcher, out_buf, r,
- FALSE, FALSE)) {
- return FALSE;
- }
- return FALSE;
- }
-
- if (!rspamd_learn_task (session->learn_symbol, task, &err)) {
- rspamd_task_free (task, FALSE);
- if (err) {
- if (session->restful) {
- i = rspamd_snprintf (out_buf,
- sizeof (out_buf),
- "HTTP/1.0 500 Learn classifier error: %s" CRLF CRLF,
- err->message);
- }
- else {
- i = rspamd_snprintf (out_buf,
- sizeof (out_buf),
- "learn failed, learn classifier error: %s" CRLF END,
- err->message);
- }
- g_error_free (err);
- }
- else {
- if (session->restful) {
- i = rspamd_snprintf (out_buf,
- sizeof (out_buf),
- "HTTP/1.0 500 Learn classifier error: unknown" CRLF CRLF);
- }
- else {
- i = rspamd_snprintf (out_buf,
- sizeof (out_buf),
- "learn failed, unknown learn classifier error" CRLF END);
- }
- }
-
- if (!rspamd_dispatcher_write (session->dispatcher, out_buf, i,
- FALSE, FALSE)) {
- return FALSE;
- }
- session->state = STATE_REPLY;
- return TRUE;
- }
-
- rspamd_task_free (task, FALSE);
- if (session->restful) {
- i = rspamd_snprintf (out_buf,
- sizeof (out_buf),
- "HTTP/1.0 200 Learn OK" CRLF CRLF);
- }
- else {
- i =
- rspamd_snprintf (out_buf, sizeof (out_buf),
- "learn ok" CRLF END);
- }
- session->state = STATE_REPLY;
- if (!rspamd_dispatcher_write (session->dispatcher, out_buf, i, FALSE,
- FALSE)) {
- return FALSE;
- }
- break;
- case STATE_LEARN_SPAM_PRE:
- session->learn_buf = in;
- task = rspamd_task_new (session->worker);
-
- task->msg = g_string_new_len (in->begin, in->len);
-
- task->resolver = session->resolver;
- task->ev_base = session->ev_base;
-
- r = process_message (task);
- if (r == -1) {
- msg_warn ("processing of message failed");
- session->state = STATE_REPLY;
- if (session->restful) {
- r = rspamd_snprintf (out_buf,
- sizeof (out_buf),
- "HTTP/1.0 500 Cannot process message" CRLF CRLF);
- }
- else {
- r = rspamd_snprintf (out_buf,
- sizeof (out_buf),
- "cannot process message" CRLF);
- }
- if (!session->restful) {
- if (!rspamd_dispatcher_write (session->dispatcher, out_buf, r,
- FALSE, FALSE)) {
- return FALSE;
- }
- }
- else {
- if (!restful_write_reply (500, out_buf, NULL, 0,
- session->dispatcher)) {
- return FALSE;
- }
- }
- return FALSE;
- }
- /* Set up async session */
- task->s = new_async_session (task->task_pool,
- fin_learn_task,
- restore_learn_task,
- rspamd_task_free_hard,
- task);
- session->learn_task = task;
- session->state = STATE_LEARN_SPAM;
- rspamd_dispatcher_pause (session->dispatcher);
- r = rspamd_process_filters (task);
- if (r == -1) {
- rspamd_dispatcher_restore (session->dispatcher);
- session->state = STATE_REPLY;
- if (session->restful) {
- r = rspamd_snprintf (out_buf,
- sizeof (out_buf),
- "HTTP/1.0 500 Cannot process message" CRLF CRLF);
- }
- else {
- r = rspamd_snprintf (out_buf,
- sizeof (out_buf),
- "cannot process message" CRLF);
- }
- destroy_session (task->s);
- if (!session->restful) {
- if (!rspamd_dispatcher_write (session->dispatcher, out_buf, r,
- FALSE, FALSE)) {
- return FALSE;
- }
- }
- else {
- if (!restful_write_reply (500, out_buf, NULL, 0,
- session->dispatcher)) {
- return FALSE;
- }
- }
- }
- break;
- case STATE_WEIGHTS:
- session->learn_buf = in;
- task = rspamd_task_new (session->worker);
-
- task->msg = g_string_new_len (in->begin, in->len);
- task->ev_base = session->ev_base;
-
- r = process_message (task);
- if (r == -1) {
- msg_warn ("processing of message failed");
- rspamd_task_free (task, FALSE);
- session->state = STATE_REPLY;
- r = rspamd_snprintf (out_buf,
- sizeof (out_buf),
- "cannot process message" CRLF END);
- if (!session->restful) {
- if (!rspamd_dispatcher_write (session->dispatcher, out_buf, r,
- FALSE, FALSE)) {
- return FALSE;
- }
- }
- else {
- if (!restful_write_reply (500, out_buf, NULL, 0,
- session->dispatcher)) {
- return FALSE;
- }
- }
- return FALSE;
- }
-
- cur = g_list_first (task->text_parts);
- while (cur) {
- part = cur->data;
- if (part->is_empty) {
- cur = g_list_next (cur);
- continue;
- }
-
- c.begin = part->content->data;
- c.len = part->content->len;
- if (!session->learn_classifier->tokenizer->tokenize_func (session->
- learn_classifier->tokenizer,
- session->session_pool, &c, &tokens, FALSE, part->is_utf,
- part->urls_offset)) {
- i = rspamd_snprintf (out_buf,
- sizeof (out_buf),
- "weights failed, tokenizer error" CRLF END);
- rspamd_task_free (task, FALSE);
- if (!session->restful) {
- if (!rspamd_dispatcher_write (session->dispatcher, out_buf,
- r, FALSE, FALSE)) {
- return FALSE;
- }
- }
- else {
- if (!restful_write_reply (500, out_buf, NULL, 0,
- session->dispatcher)) {
- return FALSE;
- }
- }
- session->state = STATE_REPLY;
- return TRUE;
- }
- cur = g_list_next (cur);
- }
-
- /* Handle messages without text */
- if (tokens == NULL) {
- i = rspamd_snprintf (out_buf,
- sizeof (out_buf),
- "weights failed, no tokens can be extracted (no text data)" CRLF END);
- rspamd_task_free (task, FALSE);
- if (!rspamd_dispatcher_write (session->dispatcher, out_buf, i,
- FALSE, FALSE)) {
- return FALSE;
- }
- session->state = STATE_REPLY;
- return TRUE;
- }
-
-
- /* Init classifier */
- cls_ctx = session->learn_classifier->classifier->init_func (
- session->session_pool,
- session->learn_classifier);
-
- cur = session->learn_classifier->classifier->weights_func (cls_ctx,
- session->worker->srv->statfile_pool,
- tokens,
- task);
- i = 0;
- struct classify_weight *w;
-
- while (cur) {
- w = cur->data;
- i += rspamd_snprintf (out_buf + i,
- sizeof (out_buf) - i,
- "%s: %G" CRLF,
- w->name,
- w->weight);
- cur = g_list_next (cur);
- }
- i += rspamd_snprintf (out_buf + i, sizeof (out_buf) - i, END);
- if (i != 0) {
- if (!rspamd_dispatcher_write (session->dispatcher, out_buf, i,
- FALSE, FALSE)) {
- return FALSE;
- }
- }
- else {
- if (!rspamd_dispatcher_write (session->dispatcher,
- "weights failed: classifier error" CRLF END, 0, FALSE, TRUE)) {
- return FALSE;
- }
- }
-
- rspamd_task_free (task, FALSE);
-
- session->state = STATE_REPLY;
- break;
- case STATE_OTHER:
- rspamd_dispatcher_pause (session->dispatcher);
- if (session->other_handler) {
- if (!session->other_handler (session, in)) {
- return FALSE;
- }
- }
- break;
- case STATE_WAIT:
- rspamd_dispatcher_pause (session->dispatcher);
- break;
- default:
- msg_debug ("unknown state while reading %d", session->state);
- break;
- }
-
- if (session->state == STATE_REPLY || session->state == STATE_QUIT) {
- /* In case of normal session we restore read state, for restful session we need to terminate immediately */
- if (!session->restful) {
- rspamd_dispatcher_restore (session->dispatcher);
- }
- else {
- return FALSE;
- }
- }
- return TRUE;
-}
-
-static gboolean
-controller_write_socket (void *arg)
-{
- struct controller_session *session = (struct controller_session *)arg;
- gint i;
- gchar out_buf[1024];
- GError *err = NULL;
-
- if (session->state == STATE_QUIT) {
- /* Free buffers */
- destroy_session (session->s);
- return FALSE;
- }
- else if (session->state == STATE_LEARN_SPAM) {
- /* Perform actual learn here */
- if (session->learn_classifier == NULL) {
- if (session->restful) {
- i = rspamd_snprintf (out_buf,
- sizeof (out_buf),
- "HTTP/1.0 500 Learn classifier error: %s" CRLF CRLF,
- "unknown classifier");
- }
- else {
- i = rspamd_snprintf (out_buf,
- sizeof (out_buf),
- "learn failed, learn classifier error: %s" CRLF END,
- "unknown classifier");
- }
- }
- else {
- if (!rspamd_learn_task_spam (session->learn_classifier,
- session->learn_task, session->in_class, &err)) {
- if (err) {
- if (session->restful) {
- i = rspamd_snprintf (out_buf,
- sizeof (out_buf),
- "HTTP/1.0 500 Learn classifier error: %s" CRLF CRLF,
- err->message);
- }
- else {
- i = rspamd_snprintf (out_buf,
- sizeof (out_buf),
- "learn failed, learn classifier error: %s" CRLF END,
- err->message);
- }
- g_error_free (err);
- }
- else {
- if (session->restful) {
- i = rspamd_snprintf (out_buf,
- sizeof (out_buf),
- "HTTP/1.0 500 Learn classifier error: unknown" CRLF CRLF);
- }
- else {
- i = rspamd_snprintf (out_buf,
- sizeof (out_buf),
- "learn failed, learn classifier error: unknown" CRLF END);
- }
- }
- }
- else {
- if (session->restful) {
- i = rspamd_snprintf (out_buf,
- sizeof (out_buf),
- "HTTP/1.0 200 Learn OK" CRLF CRLF);
- }
- else {
- i = rspamd_snprintf (out_buf,
- sizeof (out_buf),
- "learn ok" CRLF END);
- }
- }
- }
- destroy_session (session->learn_task->s);
- session->state = STATE_REPLY;
- if (!rspamd_dispatcher_write (session->dispatcher, out_buf, i, FALSE,
- FALSE)) {
- return FALSE;
- }
- if (session->restful) {
- destroy_session (session->s);
- return FALSE;
- }
- return TRUE;
- }
- else if (session->state == STATE_REPLY) {
- if (session->restful) {
- destroy_session (session->s);
- return FALSE;
- }
- else {
- session->state = STATE_COMMAND;
- rspamd_set_dispatcher_policy (session->dispatcher,
- BUFFER_LINE,
- BUFSIZ);
- }
- }
- rspamd_dispatcher_restore (session->dispatcher);
- return TRUE;
-}
-
-static void
-controller_err_socket (GError * err, void *arg)
-{
- struct controller_session *session = (struct controller_session *)arg;
-
- if (err->code != EOF) {
- msg_info ("abnormally closing control connection, error: %s",
- err->message);
- }
- g_error_free (err);
- /* Free buffers */
- destroy_session (session->s);
-}
-
-static void
-accept_socket (gint fd, short what, void *arg)
-{
- struct rspamd_worker *worker = (struct rspamd_worker *)arg;
- union sa_union su;
- struct controller_session *new_session;
- struct timeval *io_tv;
- gint nfd;
- struct rspamd_controller_ctx *ctx;
- rspamd_inet_addr_t addr;
-
- ctx = worker->ctx;
-
- if ((nfd = rspamd_accept_from_socket (fd, &addr)) == -1) {
- return;
- }
-
- new_session = g_slice_alloc0 (sizeof (struct controller_session));
- if (new_session == NULL) {
- msg_err ("cannot allocate memory for task, %s", strerror (errno));
- return;
- }
-
- new_session->worker = worker;
- new_session->sock = nfd;
- new_session->cfg = worker->srv->cfg;
- new_session->state = STATE_COMMAND;
- new_session->session_pool = rspamd_mempool_new (
- rspamd_mempool_suggest_size () - 1);
- new_session->resolver = ctx->resolver;
- new_session->ev_base = ctx->ev_base;
- if (ctx->password == NULL) {
- new_session->authorized = TRUE;
- }
- worker->srv->stat->control_connections_count++;
-
- /* Set up dispatcher */
- io_tv =
- rspamd_mempool_alloc (new_session->session_pool,
- sizeof (struct timeval));
- io_tv->tv_sec = ctx->timeout / 1000;
- io_tv->tv_usec = ctx->timeout - io_tv->tv_sec * 1000;
-
- new_session->s = new_async_session (new_session->session_pool,
- NULL,
- NULL,
- free_session,
- new_session);
-
- new_session->dispatcher = rspamd_create_dispatcher (ctx->ev_base,
- nfd,
- BUFFER_LINE,
- controller_read_socket,
- controller_write_socket,
- controller_err_socket,
- io_tv,
- (void *)new_session);
-
- if (su.ss.ss_family == AF_UNIX) {
- msg_info ("accepted connection from unix socket");
- new_session->dispatcher->peer_addr = INADDR_LOOPBACK;
- }
- else if (su.ss.ss_family == AF_INET) {
- msg_info ("accepted connection from %s port %d",
- inet_ntoa (su.s4.sin_addr), ntohs (su.s4.sin_port));
- memcpy (&new_session->dispatcher->peer_addr, &su.s4.sin_addr,
- sizeof (guint32));
- }
-#if 0
- /* As we support now restful HTTP like connections, this should not be printed anymore */
- if (!rspamd_dispatcher_write (new_session->dispatcher, greetingbuf,
- strlen (greetingbuf), FALSE, FALSE)) {
- msg_warn ("cannot write greeting");
- }
-#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 (struct rspamd_config *cfg)
-{
- struct rspamd_controller_ctx *ctx;
- GQuark type;
-
- type = g_quark_try_string ("controller");
- ctx = g_malloc0 (sizeof (struct rspamd_controller_ctx));
-
- ctx->timeout = CONTROLLER_IO_TIMEOUT * 1000;
-
- rspamd_rcl_register_worker_option (cfg, type, "password",
- rspamd_rcl_parse_struct_string, ctx,
- G_STRUCT_OFFSET (struct rspamd_controller_ctx, password), 0);
-
- rspamd_rcl_register_worker_option (cfg, type, "timeout",
- rspamd_rcl_parse_struct_time, ctx,
- G_STRUCT_OFFSET (struct rspamd_controller_ctx,
- timeout), RSPAMD_CL_FLAG_TIME_UINT_32);
-
- return ctx;
-}
-
-void
-start_controller (struct rspamd_worker *worker)
-{
- gchar *hostbuf;
- gsize hostmax;
- struct rspamd_controller_ctx *ctx = worker->ctx;
- GError *err = NULL;
- struct timeval tv;
-
- ctx->ev_base = rspamd_prepare_worker (worker, "controller", accept_socket);
- g_mime_init (0);
-
- start_time = time (NULL);
-
- /* Start statfile synchronization */
- if (!start_statfile_sync (worker->srv->statfile_pool, worker->srv->cfg,
- ctx->ev_base)) {
- 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);
- gethostname (hostbuf, hostmax);
- hostbuf[hostmax - 1] = '\0';
- rspamd_snprintf (greetingbuf,
- sizeof (greetingbuf),
- "Rspamd version %s is running on %s" CRLF,
- RVERSION,
- hostbuf);
-
- start_map_watch (worker->srv->cfg, ctx->ev_base);
- ctx->resolver = dns_resolver_init (worker->srv->logger,
- ctx->ev_base,
- worker->srv->cfg);
-
- 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);
-}
-
-void
-register_custom_controller_command (const gchar *name,
- controller_func_t handler,
- gboolean privilleged,
- gboolean require_message)
-{
- struct custom_controller_command *cmd;
-
- cmd = g_malloc (sizeof (struct custom_controller_command));
- cmd->command = name;
- cmd->handler = handler;
- cmd->privilleged = privilleged;
- cmd->require_message = require_message;
-
- custom_commands = g_list_prepend (custom_commands, cmd);
-}
-
-/*
- * vi:ts=4
- */
projects / rspamd.git / commitdiff