<commit_before>/*-
 * Copyright 2016 <NAME>
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
/*
 * Implementation of map files handling
 */
#include "config.h"
#include "map.h"
#include "map_private.h"
#include "http.h"
#include "rspamd.h"
#include "cryptobox.h"
#include "unix-std.h"
#include "http_parser.h"

static const gchar *hash_fill = "1";
static void free_http_cbdata (struct http_callback_data *cbd);
/**
 * Write HTTP request
 */
static void
write_http_request (struct http_callback_data *cbd)
{
	gchar datebuf[128];
	struct rspamd_http_message *msg;
	rspamd_mempool_t *pool;

	pool = cbd->map->pool;

	if (cbd->fd != -1) {
		close (cbd->fd);
	}

	cbd->fd = rspamd_inet_address_connect (cbd->addr, SOCK_STREAM, TRUE);

	if (cbd->fd != -1) {
		msg = rspamd_http_new_message (HTTP_REQUEST);

		if (cbd->stage == map_load_file) {
			msg->url = rspamd_fstring_new_init (cbd->data->path, strlen (cbd->data->path));

			if (cbd->data->last_checked != 0 && cbd->stage == map_load_file) {
				rspamd_http_date_format (datebuf, sizeof (datebuf),
						cbd->data->last_checked);
				rspamd_http_message_add_header (msg, "If-Modified-Since", datebuf);
			}
		}
		else if (cbd->stage == map_load_pubkey) {
			msg->url = rspamd_fstring_new_init (cbd->data->path, strlen (cbd->data->path));
			msg->url = rspamd_fstring_append (msg->url, ".pub", 4);
		}
		else if (cbd->stage == map_load_signature) {
			msg->url = rspamd_fstring_new_init (cbd->data->path, strlen (cbd->data->path));
			msg->url = rspamd_fstring_append (msg->url, ".sig", 4);
		}

		rspamd_http_connection_write_message (cbd->conn, msg, cbd->data->host,
				NULL, cbd, cbd->fd, &cbd->tv, cbd->ev_base);
	}
	else {
		msg_err_pool ("cannot connect to %s: %s", cbd->data->host,
				strerror (errno));
	}
}

static gboolean
rspamd_map_check_sig_pk (const char *fname,
		struct rspamd_map *map,
		const guchar *input,
		gsize inlen,
		struct rspamd_cryptobox_pubkey *pk)
{
	gchar fpath[PATH_MAX];
	rspamd_mempool_t *pool = map->pool;
	guchar *data;
	GString *b32_key;
	gsize len = 0;

	/* Now load signature */
	rspamd_snprintf (fpath, sizeof (fpath), "%s.sig", fname);
	data = rspamd_file_xmap (fpath, PROT_READ, &len);

	if (data == NULL) {
		msg_err_pool ("can't open signature %s: %s", fpath, strerror (errno));
		return FALSE;
	}

	if (len != rspamd_cryptobox_signature_bytes (RSPAMD_CRYPTOBOX_MODE_25519)) {
		msg_err_pool ("can't open signature %s: invalid signature", fpath);
		munmap (data, len);

		return FALSE;
	}

	if (!rspamd_cryptobox_verify (data, input, inlen,
			rspamd_pubkey_get_pk (pk, NULL), RSPAMD_CRYPTOBOX_MODE_25519)) {
		msg_err_pool ("can't verify signature %s: incorrect signature", fpath);
		munmap (data, len);

		return FALSE;
	}

	b32_key = rspamd_pubkey_print (pk,
			RSPAMD_KEYPAIR_BASE32|RSPAMD_KEYPAIR_PUBKEY);
	msg_info_pool ("verified signature in file %s using trusted key %v",
			fpath, b32_key);
	g_string_free (b32_key, TRUE);

	munmap (data, len);

	return TRUE;
}

static gboolean
rspamd_map_check_file_sig (const char *fname,
		struct rspamd_map *map, const guchar *input,
		gsize inlen)
{
	gchar fpath[PATH_MAX];
	rspamd_mempool_t *pool = map->pool;
	guchar *data;
	struct rspamd_cryptobox_pubkey *pk = NULL;
	GString *b32_key;
	gboolean ret;
	gsize len = 0;

	if (map->trusted_pubkey == NULL) {
		/* Try to load and check pubkey */
		rspamd_snprintf (fpath, sizeof (fpath), "%s.pub", fname);

		data = rspamd_file_xmap (fpath, PROT_READ, &len);

		if (data == NULL) {
			msg_err_pool ("can't open pubkey %s: %s", fpath, strerror (errno));
			return FALSE;
		}

		pk = rspamd_pubkey_from_base32 (data, len, RSPAMD_KEYPAIR_SIGN,
				RSPAMD_CRYPTOBOX_MODE_25519);
		munmap (data, len);

		if (pk == NULL) {
			msg_err_pool ("can't load pubkey %s", fpath);
			return FALSE;
		}

		/* We just check pk against the trusted db of keys */
		b32_key = rspamd_pubkey_print (pk,
				RSPAMD_KEYPAIR_BASE32|RSPAMD_KEYPAIR_PUBKEY);
		g_assert (b32_key != NULL);

		if (g_hash_table_lookup (map->cfg->trusted_keys, b32_key->str) == NULL) {
			msg_err_pool ("pubkey loaded from %s is untrusted: %v", fpath,
					b32_key);
			g_string_free (b32_key, TRUE);
			rspamd_pubkey_unref (pk);

			return FALSE;
		}

		g_string_free (b32_key, TRUE);
	}
	else {
		pk = rspamd_pubkey_ref (map->trusted_pubkey);
	}

	ret = rspamd_map_check_sig_pk (fname, map, input, inlen, pk);
	rspamd_pubkey_unref (pk);

	return ret;
}

/**
 * Callback for destroying HTTP callback data
 */
static void
free_http_cbdata (struct http_callback_data *cbd)
{
	char fpath[PATH_MAX];
	struct stat st;

	if (cbd->out_fd != -1) {
		close (cbd->out_fd);
	}

	rspamd_snprintf (fpath, sizeof (fpath), "%s", cbd->tmpfile);
	if (stat (fpath, &st) != -1 && S_ISREG (st.st_mode)) {
		(void)unlink (fpath);
	}

	rspamd_snprintf (fpath, sizeof (fpath), "%s.pub", cbd->tmpfile);
	if (stat (fpath, &st) != -1 && S_ISREG (st.st_mode)) {
		(void)unlink (fpath);
	}

	rspamd_snprintf (fpath, sizeof (fpath), "%s.sig", cbd->tmpfile);
	if (stat (fpath, &st) != -1 && S_ISREG (st.st_mode)) {
		(void)unlink (fpath);
	}

	if (cbd->pk) {
		rspamd_pubkey_unref (cbd->pk);
	}

	if (cbd->conn) {
		rspamd_http_connection_free (cbd->conn);
	}

	if (cbd->fd != -1) {
		close (cbd->fd);
	}

	g_slice_free1 (sizeof (struct http_callback_data), cbd);
}

/*
 * HTTP callbacks
 */
static void
http_map_error (struct rspamd_http_connection *conn,
	GError *err)
{
	struct http_callback_data *cbd = conn->ud;
	rspamd_mempool_t *pool;

	pool = cbd->map->pool;

	msg_err_pool ("connection with http server terminated incorrectly: %s",
			err->message);
	free_http_cbdata (cbd);
}

static int
http_map_finish (struct rspamd_http_connection *conn,
		struct rspamd_http_message *msg)
{
	struct http_callback_data *cbd = conn->ud;
	struct rspamd_map *map;
	rspamd_mempool_t *pool;
	char fpath[PATH_MAX];
	guchar *aux_data, *in = NULL;
	gsize inlen = 0;
	struct stat st;

	map = cbd->map;
	pool = cbd->map->pool;

	if (msg->code == 200) {

		if (cbd->stage == map_load_file) {
			if (msg->last_modified) {
				cbd->data->last_checked = msg->last_modified;
			}
			else {
				cbd->data->last_checked = msg->date;
			}

			/* Maybe we need to check signature ? */
			if (map->is_signed) {
				close (cbd->out_fd);

				if (map->trusted_pubkey) {
					/* No need to load key */
					cbd->stage = map_load_signature;
					cbd->pk = rspamd_pubkey_ref (map->trusted_pubkey);
					rspamd_snprintf (fpath, sizeof (fpath), "%s.sig",
							cbd->tmpfile);
				}
				else {
					rspamd_snprintf (fpath, sizeof (fpath), "%s.pub",
							cbd->tmpfile);
					cbd->stage = map_load_pubkey;
				}

				cbd->out_fd = rspamd_file_xopen (fpath, O_RDWR|O_CREAT, 00644);

				if (cbd->out_fd == -1) {
					msg_err_pool ("cannot open pubkey file %s for writing: %s",
							fpath, strerror (errno));
					free_http_cbdata (cbd);

					return 0;
				}

				rspamd_http_connection_reset (cbd->conn);
				write_http_request (cbd);

				return 0;
			}
			else {
				/* Unsinged version - just open file */
				in = rspamd_file_xmap (cbd->tmpfile, PROT_READ, &inlen);

				if (in == NULL) {
					msg_err_pool ("cannot read tempfile %s: %s", cbd->tmpfile,
							strerror (errno));
					free_http_cbdata (cbd);

					return 0;
				}
			}
		}
		else if (cbd->stage == map_load_pubkey) {
			/* We now can load pubkey */
			(void)lseek (cbd->out_fd, 0, SEEK_SET);

			if (fstat (cbd->out_fd, &st) == -1) {
				msg_err_pool ("cannot stat pubkey file %s: %s",
						fpath, strerror (errno));
				free_http_cbdata (cbd);

				return 0;
			}

			aux_data = mmap (NULL, st.st_size, PROT_READ, MAP_SHARED,
					cbd->out_fd, 0);
			close (cbd->out_fd);
			cbd->out_fd = -1;

			if (aux_data == MAP_FAILED) {
				msg_err_pool ("cannot map pubkey file %s: %s",
						fpath, strerror (errno));
				free_http_cbdata (cbd);

				return 0;
			}

			cbd->pk = rspamd_pubkey_from_base32 (aux_data, st.st_size,
					RSPAMD_KEYPAIR_SIGN, RSPAMD_CRYPTOBOX_MODE_25519);
			munmap (aux_data, st.st_size);

			if (cbd->pk == NULL) {
				msg_err_pool ("cannot load pubkey file %s: bad pubkey",
						fpath);
				free_http_cbdata (cbd);

				return 0;
			}

			rspamd_snprintf (fpath, sizeof (fpath), "%s.sig", cbd->tmpfile);
			cbd->out_fd = rspamd_file_xopen (fpath, O_RDWR|O_CREAT, 00644);

			if (cbd->out_fd == -1) {
				msg_err_pool ("cannot open signature file %s for writing: %s",
						fpath, strerror (errno));
				free_http_cbdata (cbd);

				return 0;
			}

			cbd->stage = map_load_signature;
			rspamd_http_connection_reset (cbd->conn);
			write_http_request (cbd);

			return 0;
		}
		else if (cbd->stage == map_load_signature) {
			/* We can now check signature */
			close (cbd->out_fd);
			cbd->out_fd = -1;

			in = rspamd_file_xmap (cbd->tmpfile, PROT_READ, &inlen);

			if (in == NULL) {
				msg_err_pool ("cannot read tempfile %s: %s", cbd->tmpfile,
						strerror (errno));
				free_http_cbdata (cbd);

				return 0;
			}

			if (!rspamd_map_check_sig_pk (cbd->tmpfile, map, in, inlen, cbd->pk)) {
				free_http_cbdata (cbd);

				return 0;
			}
		}

		g_assert (in != NULL);

		map->read_callback (map->pool, in, inlen, &cbd->cbdata);
		map->fin_callback (map->pool, &cbd->cbdata);

		*map->user_data = cbd->cbdata.cur_data;
		msg_info_pool ("read map data from %s", cbd->data->host);
	}
	else if (msg->code == 304 && cbd->stage == map_load_file) {
		msg_debug_pool ("data is not modified for server %s",
				cbd->data->host);

		if (msg->last_modified) {
			cbd->data->last_checked = msg->last_modified;
		}
		else {
			cbd->data->last_checked = msg->date;
		}
	}
	else {
		msg_info_pool ("cannot load map %s from %s: HTTP error %d",
				map->uri, cbd->data->host, msg->code);
	}

	free_http_cbdata (cbd);

	return 0;
}

static int
http_map_read (struct rspamd_http_connection *conn,
	struct rspamd_http_message *msg,
	const gchar *chunk,
	gsize len)
{
	struct http_callback_data *cbd = conn->ud;
	rspamd_mempool_t *pool;

	if (msg->code != 200 || len == 0) {
		/* Ignore not full replies */
		return 0;
	}

	pool = cbd->map->pool;

	if (write (cbd->out_fd, chunk, len) == -1) {
		msg_err_pool ("cannot write to %s: %s", cbd->tmpfile, strerror (errno));
		free_http_cbdata (cbd);

		return -1;
	}

	return 0;
}

/**
 * Callback for reading data from file
 */
static void
read_map_file (struct rspamd_map *map, struct file_map_data *data)
{
	struct map_cb_data cbdata;
	guchar *bytes;
	gsize len;
	rspamd_mempool_t *pool = map->pool;

	if (map->read_callback == NULL || map->fin_callback == NULL) {
		msg_err_pool ("bad callback for reading map file");
		return;
	}

	bytes = rspamd_file_xmap (data->filename, PROT_READ, &len);

	if (bytes == NULL) {
		msg_err_pool ("can't open map %s: %s", data->filename, strerror (errno));
		return;
	}

	cbdata.state = 0;
	cbdata.prev_data = *map->user_data;
	cbdata.cur_data = NULL;
	cbdata.map = map;

	if (map->is_signed) {
		if (!rspamd_map_check_file_sig (data->filename, map, bytes, len)) {
			munmap (bytes, len);

			return;
		}
	}

	map->read_callback (map->pool, bytes, len, &cbdata);

	if (len > 0) {
		map->fin_callback (map->pool, &cbdata);
		*map->user_data = cbdata.cur_data;
	}

	munmap (bytes, len);
}

static void
jitter_timeout_event (struct rspamd_map *map, gboolean locked, gboolean initial)
{
	gdouble jittered_sec;
	gdouble timeout = initial ? 1.0 : map->cfg->map_timeout;

	/* Plan event again with jitter */
	evtimer_del (&map->ev);
	jittered_sec = rspamd_time_jitter (locked ? timeout * 4 : timeout, 0);
	double_to_tv (jittered_sec, &map->tv);

	evtimer_add (&map->ev, &map->tv);
}

/**
 * Common file callback
 */
static void
file_callback (gint fd, short what, void *ud)
{
	struct rspamd_map *map = ud;
	struct file_map_data *data = map->map_data;
	struct stat st;
	rspamd_mempool_t *pool;

	pool = map->pool;

	if (g_atomic_int_get (map->locked)) {
		msg_info_pool (
			"don't try to reread map as it is locked by other process, will reread it later");
		jitter_timeout_event (map, TRUE, FALSE);
		return;
	}

	g_atomic_int_inc (map->locked);
	jitter_timeout_event (map, FALSE, FALSE);
	if (stat (data->filename,
		&st) != -1 &&
		(st.st_mtime > data->st.st_mtime || data->st.st_mtime == -1)) {
		/* File was modified since last check */
		memcpy (&data->st, &st, sizeof (struct stat));
	}
	else {
		g_atomic_int_set (map->locked, 0);
		return;
	}

	msg_info_pool ("rereading map file %s", data->filename);
	read_map_file (map, data);
	g_atomic_int_set (map->locked, 0);
}


static void
rspamd_map_dns_callback (struct rdns_reply *reply, void *arg)
{
	struct http_callback_data *cbd = arg;
	rspamd_mempool_t *pool;

	if (cbd->stage >= map_load_file) {
		/* No need in further corrections */
		return;
	}

	pool = cbd->map->pool;

	if (reply->code == RDNS_RC_NOERROR) {
		/*
		 * We just get the first address hoping that a resolver performs
		 * round-robin rotation well
		 */
		cbd->addr = rspamd_inet_address_from_rnds (reply->entries);


		if (cbd->addr != NULL) {
			rspamd_inet_address_set_port (cbd->addr, cbd->data->port);
			/* Try to open a socket */

			cbd->fd = rspamd_inet_address_connect (cbd->addr, SOCK_STREAM, TRUE);

			if (cbd->fd != -1) {
				cbd->stage = map_load_file;
				cbd->conn = rspamd_http_connection_new (http_map_read,
						http_map_error, http_map_finish,
						RSPAMD_HTTP_BODY_PARTIAL|RSPAMD_HTTP_CLIENT_SIMPLE,
						RSPAMD_HTTP_CLIENT, NULL);

				write_http_request (cbd);
			}
		}
	}

	if (cbd->stage < map_load_file) {
		if (cbd->stage == map_resolve_host2) {
			/* We have still one request pending */
			cbd->stage = map_resolve_host1;
		}
		else {
			/* We could not resolve host, so cowardly fail here */
			msg_err_pool ("cannot resolve %s", cbd->data->host);
			free_http_cbdata (cbd);
		}
	}
}

/**
 * Async HTTP callback
 */
static void
http_callback (gint fd, short what, void *ud)
{
	struct rspamd_map *map = ud;
	struct http_map_data *data;
	struct http_callback_data *cbd;
	rspamd_mempool_t *pool;
	gchar tmpbuf[PATH_MAX];

	data = map->map_data;
	pool = map->pool;

	jitter_timeout_event (map, FALSE, FALSE);
	/* Plan event */
	cbd = g_slice_alloc0 (sizeof (struct http_callback_data));

	rspamd_snprintf (tmpbuf, sizeof (tmpbuf),
			"%s" G_DIR_SEPARATOR_S "rspamd_map%d-XXXXXX",
			map->cfg->temp_dir, map->id);
	cbd->out_fd = mkstemp (tmpbuf);

	if (cbd->out_fd == -1) {
		g_slice_free1 (sizeof (*cbd), cbd);
		msg_err_pool ("cannot create tempfile: %s", strerror (errno));
		return;
	}

	cbd->tmpfile = g_strdup (tmpbuf);
	cbd->ev_base = map->ev_base;
	cbd->map = map;
	cbd->data = data;
	cbd->fd = -1;
	cbd->cbdata.state = 0;
	cbd->cbdata.prev_data = *cbd->map->user_data;
	cbd->cbdata.cur_data = NULL;
	cbd->cbdata.map = cbd->map;
	cbd->stage = map_resolve_host2;
	double_to_tv (map->cfg->map_timeout, &cbd->tv);

	msg_debug_pool ("reading map data from %s", data->host);
	/* Send both A and AAAA requests */
	if (map->r->r) {
		rdns_make_request_full (map->r->r, rspamd_map_dns_callback, cbd,
				map->cfg->dns_timeout, map->cfg->dns_retransmits, 1,
				data->host, RDNS_REQUEST_A);
		rdns_make_request_full (map->r->r, rspamd_map_dns_callback, cbd,
				map->cfg->dns_timeout, map->cfg->dns_retransmits, 1,
				data->host, RDNS_REQUEST_AAAA);
	}
	else {
		msg_warn_pool ("cannot load map: DNS resolver is not initialized");
		free_http_cbdata (cbd);
	}
}

/* Start watching event for all maps */
void
rspamd_map_watch (struct rspamd_config *cfg,
		struct event_base *ev_base,
		struct rspamd_dns_resolver *resolver)
{
	GList *cur = cfg->maps;
	struct rspamd_map *map;
	struct file_map_data *fdata;

	/* First of all do synced read of data */
	while (cur) {
		map = cur->data;
		map->ev_base = ev_base;
		map->r = resolver;
		event_base_set (map->ev_base, &map->ev);

		if (map->protocol == MAP_PROTO_FILE) {
			evtimer_set (&map->ev, file_callback, map);
			/* Read initial data */
			fdata = map->map_data;
			if (fdata->st.st_mtime != -1) {
				/* Do not try to read non-existent file */
				read_map_file (map, map->map_data);
			}
			/* Plan event with jitter */
			jitter_timeout_event (map, FALSE, TRUE);
		}
		else if (map->protocol == MAP_PROTO_HTTP) {
			evtimer_set (&map->ev, http_callback, map);
			jitter_timeout_event (map, FALSE, TRUE);
		}

		cur = g_list_next (cur);
	}
}

void
rspamd_map_remove_all (struct rspamd_config *cfg)<commit_msg>[Fetaure] Be more clever about map timeouts<commit_after>/*-
 * Copyright 2016 <NAME>
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
/*
 * Implementation of map files handling
 */
#include "config.h"
#include "map.h"
#include "map_private.h"
#include "http.h"
#include "rspamd.h"
#include "cryptobox.h"
#include "unix-std.h"
#include "http_parser.h"

static const gchar *hash_fill = "1";
static void free_http_cbdata (struct http_callback_data *cbd);
/**
 * Write HTTP request
 */
static void
write_http_request (struct http_callback_data *cbd)
{
	gchar datebuf[128];
	struct rspamd_http_message *msg;
	rspamd_mempool_t *pool;

	pool = cbd->map->pool;

	if (cbd->fd != -1) {
		close (cbd->fd);
	}

	cbd->fd = rspamd_inet_address_connect (cbd->addr, SOCK_STREAM, TRUE);

	if (cbd->fd != -1) {
		msg = rspamd_http_new_message (HTTP_REQUEST);

		if (cbd->stage == map_load_file) {
			msg->url = rspamd_fstring_new_init (cbd->data->path, strlen (cbd->data->path));

			if (cbd->data->last_checked != 0 && cbd->stage == map_load_file) {
				rspamd_http_date_format (datebuf, sizeof (datebuf),
						cbd->data->last_checked);
				rspamd_http_message_add_header (msg, "If-Modified-Since", datebuf);
			}
		}
		else if (cbd->stage == map_load_pubkey) {
			msg->url = rspamd_fstring_new_init (cbd->data->path, strlen (cbd->data->path));
			msg->url = rspamd_fstring_append (msg->url, ".pub", 4);
		}
		else if (cbd->stage == map_load_signature) {
			msg->url = rspamd_fstring_new_init (cbd->data->path, strlen (cbd->data->path));
			msg->url = rspamd_fstring_append (msg->url, ".sig", 4);
		}

		rspamd_http_connection_write_message (cbd->conn, msg, cbd->data->host,
				NULL, cbd, cbd->fd, &cbd->tv, cbd->ev_base);
	}
	else {
		msg_err_pool ("cannot connect to %s: %s", cbd->data->host,
				strerror (errno));
	}
}

static gboolean
rspamd_map_check_sig_pk (const char *fname,
		struct rspamd_map *map,
		const guchar *input,
		gsize inlen,
		struct rspamd_cryptobox_pubkey *pk)
{
	gchar fpath[PATH_MAX];
	rspamd_mempool_t *pool = map->pool;
	guchar *data;
	GString *b32_key;
	gsize len = 0;

	/* Now load signature */
	rspamd_snprintf (fpath, sizeof (fpath), "%s.sig", fname);
	data = rspamd_file_xmap (fpath, PROT_READ, &len);

	if (data == NULL) {
		msg_err_pool ("can't open signature %s: %s", fpath, strerror (errno));
		return FALSE;
	}

	if (len != rspamd_cryptobox_signature_bytes (RSPAMD_CRYPTOBOX_MODE_25519)) {
		msg_err_pool ("can't open signature %s: invalid signature", fpath);
		munmap (data, len);

		return FALSE;
	}

	if (!rspamd_cryptobox_verify (data, input, inlen,
			rspamd_pubkey_get_pk (pk, NULL), RSPAMD_CRYPTOBOX_MODE_25519)) {
		msg_err_pool ("can't verify signature %s: incorrect signature", fpath);
		munmap (data, len);

		return FALSE;
	}

	b32_key = rspamd_pubkey_print (pk,
			RSPAMD_KEYPAIR_BASE32|RSPAMD_KEYPAIR_PUBKEY);
	msg_info_pool ("verified signature in file %s using trusted key %v",
			fpath, b32_key);
	g_string_free (b32_key, TRUE);

	munmap (data, len);

	return TRUE;
}

static gboolean
rspamd_map_check_file_sig (const char *fname,
		struct rspamd_map *map, const guchar *input,
		gsize inlen)
{
	gchar fpath[PATH_MAX];
	rspamd_mempool_t *pool = map->pool;
	guchar *data;
	struct rspamd_cryptobox_pubkey *pk = NULL;
	GString *b32_key;
	gboolean ret;
	gsize len = 0;

	if (map->trusted_pubkey == NULL) {
		/* Try to load and check pubkey */
		rspamd_snprintf (fpath, sizeof (fpath), "%s.pub", fname);

		data = rspamd_file_xmap (fpath, PROT_READ, &len);

		if (data == NULL) {
			msg_err_pool ("can't open pubkey %s: %s", fpath, strerror (errno));
			return FALSE;
		}

		pk = rspamd_pubkey_from_base32 (data, len, RSPAMD_KEYPAIR_SIGN,
				RSPAMD_CRYPTOBOX_MODE_25519);
		munmap (data, len);

		if (pk == NULL) {
			msg_err_pool ("can't load pubkey %s", fpath);
			return FALSE;
		}

		/* We just check pk against the trusted db of keys */
		b32_key = rspamd_pubkey_print (pk,
				RSPAMD_KEYPAIR_BASE32|RSPAMD_KEYPAIR_PUBKEY);
		g_assert (b32_key != NULL);

		if (g_hash_table_lookup (map->cfg->trusted_keys, b32_key->str) == NULL) {
			msg_err_pool ("pubkey loaded from %s is untrusted: %v", fpath,
					b32_key);
			g_string_free (b32_key, TRUE);
			rspamd_pubkey_unref (pk);

			return FALSE;
		}

		g_string_free (b32_key, TRUE);
	}
	else {
		pk = rspamd_pubkey_ref (map->trusted_pubkey);
	}

	ret = rspamd_map_check_sig_pk (fname, map, input, inlen, pk);
	rspamd_pubkey_unref (pk);

	return ret;
}

/**
 * Callback for destroying HTTP callback data
 */
static void
free_http_cbdata (struct http_callback_data *cbd)
{
	char fpath[PATH_MAX];
	struct stat st;

	if (cbd->out_fd != -1) {
		close (cbd->out_fd);
	}

	rspamd_snprintf (fpath, sizeof (fpath), "%s", cbd->tmpfile);
	if (stat (fpath, &st) != -1 && S_ISREG (st.st_mode)) {
		(void)unlink (fpath);
	}

	rspamd_snprintf (fpath, sizeof (fpath), "%s.pub", cbd->tmpfile);
	if (stat (fpath, &st) != -1 && S_ISREG (st.st_mode)) {
		(void)unlink (fpath);
	}

	rspamd_snprintf (fpath, sizeof (fpath), "%s.sig", cbd->tmpfile);
	if (stat (fpath, &st) != -1 && S_ISREG (st.st_mode)) {
		(void)unlink (fpath);
	}

	if (cbd->pk) {
		rspamd_pubkey_unref (cbd->pk);
	}

	if (cbd->conn) {
		rspamd_http_connection_free (cbd->conn);
	}

	if (cbd->fd != -1) {
		close (cbd->fd);
	}

	g_slice_free1 (sizeof (struct http_callback_data), cbd);
}

/*
 * HTTP callbacks
 */
static void
http_map_error (struct rspamd_http_connection *conn,
	GError *err)
{
	struct http_callback_data *cbd = conn->ud;
	rspamd_mempool_t *pool;

	pool = cbd->map->pool;

	msg_err_pool ("connection with http server terminated incorrectly: %s",
			err->message);
	free_http_cbdata (cbd);
}

static int
http_map_finish (struct rspamd_http_connection *conn,
		struct rspamd_http_message *msg)
{
	struct http_callback_data *cbd = conn->ud;
	struct rspamd_map *map;
	rspamd_mempool_t *pool;
	char fpath[PATH_MAX];
	guchar *aux_data, *in = NULL;
	gsize inlen = 0;
	struct stat st;

	map = cbd->map;
	pool = cbd->map->pool;

	if (msg->code == 200) {

		if (cbd->stage == map_load_file) {
			if (msg->last_modified) {
				cbd->data->last_checked = msg->last_modified;
			}
			else {
				cbd->data->last_checked = msg->date;
			}

			/* Maybe we need to check signature ? */
			if (map->is_signed) {
				close (cbd->out_fd);

				if (map->trusted_pubkey) {
					/* No need to load key */
					cbd->stage = map_load_signature;
					cbd->pk = rspamd_pubkey_ref (map->trusted_pubkey);
					rspamd_snprintf (fpath, sizeof (fpath), "%s.sig",
							cbd->tmpfile);
				}
				else {
					rspamd_snprintf (fpath, sizeof (fpath), "%s.pub",
							cbd->tmpfile);
					cbd->stage = map_load_pubkey;
				}

				cbd->out_fd = rspamd_file_xopen (fpath, O_RDWR|O_CREAT, 00644);

				if (cbd->out_fd == -1) {
					msg_err_pool ("cannot open pubkey file %s for writing: %s",
							fpath, strerror (errno));
					free_http_cbdata (cbd);

					return 0;
				}

				rspamd_http_connection_reset (cbd->conn);
				write_http_request (cbd);

				return 0;
			}
			else {
				/* Unsinged version - just open file */
				in = rspamd_file_xmap (cbd->tmpfile, PROT_READ, &inlen);

				if (in == NULL) {
					msg_err_pool ("cannot read tempfile %s: %s", cbd->tmpfile,
							strerror (errno));
					free_http_cbdata (cbd);

					return 0;
				}
			}
		}
		else if (cbd->stage == map_load_pubkey) {
			/* We now can load pubkey */
			(void)lseek (cbd->out_fd, 0, SEEK_SET);

			if (fstat (cbd->out_fd, &st) == -1) {
				msg_err_pool ("cannot stat pubkey file %s: %s",
						fpath, strerror (errno));
				free_http_cbdata (cbd);

				return 0;
			}

			aux_data = mmap (NULL, st.st_size, PROT_READ, MAP_SHARED,
					cbd->out_fd, 0);
			close (cbd->out_fd);
			cbd->out_fd = -1;

			if (aux_data == MAP_FAILED) {
				msg_err_pool ("cannot map pubkey file %s: %s",
						fpath, strerror (errno));
				free_http_cbdata (cbd);

				return 0;
			}

			cbd->pk = rspamd_pubkey_from_base32 (aux_data, st.st_size,
					RSPAMD_KEYPAIR_SIGN, RSPAMD_CRYPTOBOX_MODE_25519);
			munmap (aux_data, st.st_size);

			if (cbd->pk == NULL) {
				msg_err_pool ("cannot load pubkey file %s: bad pubkey",
						fpath);
				free_http_cbdata (cbd);

				return 0;
			}

			rspamd_snprintf (fpath, sizeof (fpath), "%s.sig", cbd->tmpfile);
			cbd->out_fd = rspamd_file_xopen (fpath, O_RDWR|O_CREAT, 00644);

			if (cbd->out_fd == -1) {
				msg_err_pool ("cannot open signature file %s for writing: %s",
						fpath, strerror (errno));
				free_http_cbdata (cbd);

				return 0;
			}

			cbd->stage = map_load_signature;
			rspamd_http_connection_reset (cbd->conn);
			write_http_request (cbd);

			return 0;
		}
		else if (cbd->stage == map_load_signature) {
			/* We can now check signature */
			close (cbd->out_fd);
			cbd->out_fd = -1;

			in = rspamd_file_xmap (cbd->tmpfile, PROT_READ, &inlen);

			if (in == NULL) {
				msg_err_pool ("cannot read tempfile %s: %s", cbd->tmpfile,
						strerror (errno));
				free_http_cbdata (cbd);

				return 0;
			}

			if (!rspamd_map_check_sig_pk (cbd->tmpfile, map, in, inlen, cbd->pk)) {
				free_http_cbdata (cbd);

				return 0;
			}
		}

		g_assert (in != NULL);

		map->read_callback (map->pool, in, inlen, &cbd->cbdata);
		map->fin_callback (map->pool, &cbd->cbdata);

		*map->user_data = cbd->cbdata.cur_data;
		msg_info_pool ("read map data from %s", cbd->data->host);
	}
	else if (msg->code == 304 && cbd->stage == map_load_file) {
		msg_debug_pool ("data is not modified for server %s",
				cbd->data->host);

		if (msg->last_modified) {
			cbd->data->last_checked = msg->last_modified;
		}
		else {
			cbd->data->last_checked = msg->date;
		}
	}
	else {
		msg_info_pool ("cannot load map %s from %s: HTTP error %d",
				map->uri, cbd->data->host, msg->code);
	}

	free_http_cbdata (cbd);

	return 0;
}

static int
http_map_read (struct rspamd_http_connection *conn,
	struct rspamd_http_message *msg,
	const gchar *chunk,
	gsize len)
{
	struct http_callback_data *cbd = conn->ud;
	rspamd_mempool_t *pool;

	if (msg->code != 200 || len == 0) {
		/* Ignore not full replies */
		return 0;
	}

	pool = cbd->map->pool;

	if (write (cbd->out_fd, chunk, len) == -1) {
		msg_err_pool ("cannot write to %s: %s", cbd->tmpfile, strerror (errno));
		free_http_cbdata (cbd);

		return -1;
	}

	return 0;
}

/**
 * Callback for reading data from file
 */
static gboolean
read_map_file (struct rspamd_map *map, struct file_map_data *data)
{
	struct map_cb_data cbdata;
	guchar *bytes;
	gsize len;
	rspamd_mempool_t *pool = map->pool;

	if (map->read_callback == NULL || map->fin_callback == NULL) {
		msg_err_pool ("bad callback for reading map file");
		return FALSE;
	}

	if (access (data->filename, R_OK) == -1) {
		/* File does not exist, skipping */
		msg_err_pool ("map file is unavailable for reading");
		return FALSE;
	}

	bytes = rspamd_file_xmap (data->filename, PROT_READ, &len);

	if (bytes == NULL) {
		msg_err_pool ("can't open map %s: %s", data->filename, strerror (errno));
		return FALSE;
	}

	cbdata.state = 0;
	cbdata.prev_data = *map->user_data;
	cbdata.cur_data = NULL;
	cbdata.map = map;

	if (map->is_signed) {
		if (!rspamd_map_check_file_sig (data->filename, map, bytes, len)) {
			munmap (bytes, len);

			return FALSE;
		}
	}

	map->read_callback (map->pool, bytes, len, &cbdata);

	if (len > 0) {
		map->fin_callback (map->pool, &cbdata);
		*map->user_data = cbdata.cur_data;
	}

	munmap (bytes, len);

	return TRUE;
}

static void
jitter_timeout_event (struct rspamd_map *map,
		gboolean locked, gboolean initial, gboolean errored)
{
	const gdouble error_mult = 20.0, lock_mult = 4.0;
	gdouble jittered_sec;
	gdouble timeout;

	if (initial) {
		timeout = 0.0;
	}
	else if (errored) {
		timeout = map->cfg->map_timeout * error_mult;
	}
	else if (locked) {
		timeout = map->cfg->map_timeout * lock_mult;
	}
	else {
		timeout = map->cfg->map_timeout;
	}

	/* Plan event again with jitter */
	evtimer_del (&map->ev);
	jittered_sec = rspamd_time_jitter (timeout, 0);
	double_to_tv (jittered_sec, &map->tv);

	evtimer_add (&map->ev, &map->tv);
}

/**
 * Common file callback
 */
static void
file_callback (gint fd, short what, void *ud)
{
	struct rspamd_map *map = ud;
	struct file_map_data *data = map->map_data;
	struct stat st;
	rspamd_mempool_t *pool;

	pool = map->pool;

	if (g_atomic_int_get (map->locked)) {
		msg_info_pool (
			"don't try to reread map as it is locked by other process, will reread it later");
		jitter_timeout_event (map, TRUE, FALSE, FALSE);
		return;
	}

	g_atomic_int_inc (map->locked);

	if (stat (data->filename, &st) != -1 &&
		(st.st_mtime > data->st.st_mtime || data->st.st_mtime == -1)) {
		/* File was modified since last check */
		memcpy (&data->st, &st, sizeof (struct stat));
	}
	else {
		g_atomic_int_set (map->locked, 0);
		jitter_timeout_event (map, FALSE, FALSE, FALSE);
		return;
	}

	msg_info_pool ("rereading map file %s", data->filename);

	if (!read_map_file (map, data)) {
		jitter_timeout_event (map, FALSE, FALSE, TRUE);
	}
	else {
		jitter_timeout_event (map, FALSE, FALSE, FALSE);
	}

	g_atomic_int_set (map->locked, 0);
}


static void
rspamd_map_dns_callback (struct rdns_reply *reply, void *arg)
{
	struct http_callback_data *cbd = arg;
	rspamd_mempool_t *pool;

	if (cbd->stage >= map_load_file) {
		/* No need in further corrections */
		return;
	}

	pool = cbd->map->pool;

	if (reply->code == RDNS_RC_NOERROR) {
		/*
		 * We just get the first address hoping that a resolver performs
		 * round-robin rotation well
		 */
		cbd->addr = rspamd_inet_address_from_rnds (reply->entries);


		if (cbd->addr != NULL) {
			rspamd_inet_address_set_port (cbd->addr, cbd->data->port);
			/* Try to open a socket */

			cbd->fd = rspamd_inet_address_connect (cbd->addr, SOCK_STREAM, TRUE);

			if (cbd->fd != -1) {
				cbd->stage = map_load_file;
				cbd->conn = rspamd_http_connection_new (http_map_read,
						http_map_error, http_map_finish,
						RSPAMD_HTTP_BODY_PARTIAL|RSPAMD_HTTP_CLIENT_SIMPLE,
						RSPAMD_HTTP_CLIENT, NULL);

				write_http_request (cbd);
			}
		}
	}

	if (cbd->stage < map_load_file) {
		if (cbd->stage == map_resolve_host2) {
			/* We have still one request pending */
			cbd->stage = map_resolve_host1;
		}
		else {
			/* We could not resolve host, so cowardly fail here */
			msg_err_pool ("cannot resolve %s", cbd->data->host);
			free_http_cbdata (cbd);
		}
	}
}

/**
 * Async HTTP callback
 */
static void
http_callback (gint fd, short what, void *ud)
{
	struct rspamd_map *map = ud;
	struct http_map_data *data;
	struct http_callback_data *cbd;
	rspamd_mempool_t *pool;
	gchar tmpbuf[PATH_MAX];

	data = map->map_data;
	pool = map->pool;

	/* Plan event */
	cbd = g_slice_alloc0 (sizeof (struct http_callback_data));

	rspamd_snprintf (tmpbuf, sizeof (tmpbuf),
			"%s" G_DIR_SEPARATOR_S "rspamd_map%d-XXXXXX",
			map->cfg->temp_dir, map->id);
	cbd->out_fd = mkstemp (tmpbuf);

	if (cbd->out_fd == -1) {
		g_slice_free1 (sizeof (*cbd), cbd);
		msg_err_pool ("cannot create tempfile: %s", strerror (errno));
		jitter_timeout_event (map, FALSE, FALSE, TRUE);

		return;
	}

	cbd->tmpfile = g_strdup (tmpbuf);
	cbd->ev_base = map->ev_base;
	cbd->map = map;
	cbd->data = data;
	cbd->fd = -1;
	cbd->cbdata.state = 0;
	cbd->cbdata.prev_data = *cbd->map->user_data;
	cbd->cbdata.cur_data = NULL;
	cbd->cbdata.map = cbd->map;
	cbd->stage = map_resolve_host2;
	double_to_tv (map->cfg->map_timeout, &cbd->tv);

	msg_debug_pool ("reading map data from %s", data->host);
	/* Send both A and AAAA requests */
	if (map->r->r) {
		rdns_make_request_full (map->r->r, rspamd_map_dns_callback, cbd,
				map->cfg->dns_timeout, map->cfg->dns_retransmits, 1,
				data->host, RDNS_REQUEST_A);
		rdns_make_request_full (map->r->r, rspamd_map_dns_callback, cbd,
				map->cfg->dns_timeout, map->cfg->dns_retransmits, 1,
				data->host, RDNS_REQUEST_AAAA);
		jitter_timeout_event (map, FALSE, FALSE, FALSE);
	}
	else {
		msg_warn_pool ("cannot load map: DNS resolver is not initialized");
		free_http_cbdata (cbd);
		jitter_timeout_event (map, FALSE, FALSE, TRUE);
	}
}

/* Start watching event for all maps */
void
rspamd_map_watch (struct rspamd_config *cfg,
		struct event_base *ev_base,
		struct rspamd_dns_resolver *resolver)
{
	GList *cur = cfg->maps;
	struct rspamd_map *map;
	struct file_map_data *fdata;

	/* First of all do synced read of data */
	while (cur) {
		map = cur->data;
		map->ev_base = ev_base;
		map->r = resolver;
		event_base_set (map->ev_base, &map->ev);

		if (map->protocol == MAP_PROTO_FILE) {
			evtimer_set (&map->ev, file_callback, map);
			/* Read initial data */
			fdata = map->map_data;
			if (fdata->st.st_mtime != -1) {
				/* Do not try to read non-existent file */
				read_map_file (map, map->map_data);
			}
			/* Plan event with jitter */
			jitter_timeout_event (map, FALSE, TRUE, FALSE);
		}
		else if (map->protocol == MAP_PROTO_HTTP) {
			evtimer_set (&map->ev, http_callback, map);
			jitter_timeout_event (map, FALSE, TRUE, FALSE);
		}

		cur = g_list_next (cur);
	}
}

void
rspamd_map_remove_all (struct rspamd_config *cfg)
{'fraction_non_alphanumeric': 0.14444476854417643, 'fraction_numerical': 0.008313158125018231, 'mean_word_length': 3.7093406593406595, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 508, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '761338', 'n_tokens_mistral': 15198, 'n_tokens_neox': 13567, 'n_words': 3273}