+gpointer init_fuzzy(struct rspamd_config *cfg);

+void start_fuzzy(struct rspamd_worker *worker);

+ "fuzzy", /* Name */

+ init_fuzzy, /* Init function */

+ start_fuzzy, /* Start function */

+ RSPAMD_WORKER_HAS_SOCKET,

+ RSPAMD_WORKER_SOCKET_UDP, /* UDP socket */

+ RSPAMD_WORKER_VER /* Version info */

+ khash_t(fuzzy_key_ids_set) * default_forbidden_ids;

+ khash_t(fuzzy_key_ids_set) * weak_ids;

+ struct rspamd_fuzzy_shingle_cmd cmd; /* Can handle both shingles and non-shingles */

+ kh_int_hash_equal);

+ khash_t(fuzzy_key_flag_stat) * flags_stat;

+ khash_t(fuzzy_key_ids_set) * forbidden_ids;

+static void rspamd_fuzzy_write_reply(struct fuzzy_session *session);

+static gboolean rspamd_fuzzy_process_updates_queue(struct rspamd_fuzzy_storage_ctx *ctx,

+ const gchar *source, gboolean final);

+static gboolean rspamd_fuzzy_check_client(struct rspamd_fuzzy_storage_ctx *ctx,

+ rspamd_inet_addr_t *addr);

+static void rspamd_fuzzy_maybe_call_blacklisted(struct rspamd_fuzzy_storage_ctx *ctx,

+ rspamd_inet_addr_t *addr,

+ const gchar *reason);

+rspamd_fuzzy_check_ratelimit(struct fuzzy_session *session)

+ if (rspamd_match_radix_map_addr(session->ctx->ratelimit_whitelist,

+ session->addr) != NULL) {

+ if (rspamd_inet_address_get_af(masked) == AF_INET) {

+ rspamd_inet_address_apply_mask(masked,

+ MIN(session->ctx->leaky_bucket_mask, 32));

+ rspamd_inet_address_apply_mask(masked,

+ MIN(MAX(session->ctx->leaky_bucket_mask * 4, 64), 128));

+ elt = rspamd_lru_hash_lookup(session->ctx->ratelimit_buckets, masked,

+ (time_t) session->timestamp);

+ if (isnan(elt->cur)) {

+ msg_info("ratelimiting %s (%s), %.1f max elts",

+ rspamd_inet_address_to_string(session->addr),

+ rspamd_inet_address_to_string(masked),

+ session->ctx->leaky_bucket_burst);

+ elt->cur++; /* Allow one more request */

+ msg_err("bad address length: %d, expected to be %d", (int) slen, (int) sizeof(srv_cmd.cmd.fuzzy_blocked.addr));

+ rspamd_inet_address_free(masked);

+ elt = g_malloc(sizeof(*elt));

+ rspamd_lru_hash_insert(session->ctx->ratelimit_buckets,

+ masked,

+ elt,

+ session->timestamp,

+ session->ctx->leaky_bucket_ttl);

+rspamd_fuzzy_maybe_call_blacklisted(struct rspamd_fuzzy_storage_ctx *ctx,

+ rspamd_inet_addr_t *addr,

+ const gchar *reason)

+ lua_pushcfunction(L, &rspamd_lua_traceback);

+ err_idx = lua_gettop(L);

+ lua_rawgeti(L, LUA_REGISTRYINDEX, ctx->lua_blacklist_cbref);

+ rspamd_lua_ip_push(L, addr);

+ lua_pushstring(L, reason);

+ if ((ret = lua_pcall(L, 2, 0, err_idx)) != 0) {

+ msg_err("call to lua_blacklist_cbref "

+ "script failed (%d): %s",

+ ret, lua_tostring(L, -1));

+ lua_settop(L, 0);

+rspamd_fuzzy_check_client(struct rspamd_fuzzy_storage_ctx *ctx,

+ rspamd_inet_addr_t *addr)

+ if (rspamd_match_radix_map_addr(ctx->blocked_ips,

+ addr) != NULL) {

+ rspamd_fuzzy_maybe_call_blacklisted(ctx, addr, "blacklisted");

+rspamd_fuzzy_check_write(struct fuzzy_session *session)

+ if (rspamd_inet_address_get_af(session->addr) == AF_UNIX) {

+ if (rspamd_match_radix_map_addr(session->ctx->update_ips,

+ session->addr) == NULL) {

+ RSPAMD_KEYPAIR_COMPONENT_ID, &raw_len);

+ base32_buf, sizeof(base32_buf),

+ RSPAMD_BASE32_DEFAULT);

+ if (rspamd_match_hash_map(session->ctx->update_keys, base32_buf, encoded_len)) {

+fuzzy_key_stat_dtor(gpointer p)

+ rspamd_lru_hash_destroy(st->last_ips);

+ g_free(st);

+fuzzy_key_stat_unref(gpointer p)

+ REF_RELEASE(st);

+fuzzy_key_dtor(gpointer p)

+ REF_RELEASE(key->stat);

+ g_free(key);

+fuzzy_count_callback(guint64 count, void *ud)

+fuzzy_rl_bucket_free(gpointer p)

+ struct rspamd_leaky_bucket_elt *elt = (struct rspamd_leaky_bucket_elt *) p;

+ rspamd_inet_address_free(elt->addr);

+ g_free(elt);

+fuzzy_stat_count_callback(guint64 count, void *ud)

+ ev_timer_again(ctx->event_loop, &ctx->stat_ev);

+rspamd_fuzzy_stat_callback(EV_P_ ev_timer *w, int revents)

+ (struct rspamd_fuzzy_storage_ctx *) w->data;

+ rspamd_fuzzy_backend_count(ctx->backend, fuzzy_stat_count_callback, ctx);

+fuzzy_update_version_callback(guint64 ver, void *ud)

+rspamd_fuzzy_updates_cb(gboolean success,

+ guint nadded,

+ guint ndeleted,

+ guint nextended,

+ guint nignored,

+ void *ud)

+ msg_info("successfully updated fuzzy storage %s: %d updates in queue; "

+ "%d pending currently; "

+ "%d added; %d deleted; %d extended; %d duplicates",

+ ctx->worker->cf->bind_conf ? ctx->worker->cf->bind_conf->bind_line : "unknown",

+ cbdata->updates_pending->len,

+ ctx->updates_pending->len,

+ nadded, ndeleted, nextended, nignored);

+ fuzzy_update_version_callback, NULL);

+ msg_err("cannot commit update transaction to fuzzy backend %s, discard "

+ "%ud updates after %d retries",

+ ctx->worker->cf->bind_conf ? ctx->worker->cf->bind_conf->bind_line : "unknown",

+ msg_err("cannot commit update transaction to fuzzy backend %s; "

+ "%ud updates are still left; %ud currently pending;"

+ " %d retries remaining",

+ ctx->worker->cf->bind_conf ? ctx->worker->cf->bind_conf->bind_line : "unknown",

+ cbdata->updates_pending->data,

+ cbdata->updates_pending->len);

+ cbdata->final);

+ msg_err("cannot commit update transaction to fuzzy backend %s; "

+ "%ud updates are still left; no more retries: a worker is terminated",

+ ctx->worker->cf->bind_conf ? ctx->worker->cf->bind_conf->bind_line : "unknown",

+ g_array_free(cbdata->updates_pending, TRUE);

+ g_free(cbdata->source);

+ g_free(cbdata);

+rspamd_fuzzy_process_updates_queue(struct rspamd_fuzzy_storage_ctx *ctx,

+ const gchar *source, gboolean final)

+ cbdata = g_malloc(sizeof(*cbdata));

+ ctx->updates_pending = g_array_sized_new(FALSE, FALSE,

+ sizeof(struct fuzzy_peer_cmd),

+ MAX(cbdata->updates_pending->len, 1024));

+ cbdata->source = g_strdup(source);

+ rspamd_fuzzy_backend_process_updates(ctx->backend,

+ cbdata->updates_pending,

+ source, rspamd_fuzzy_updates_cb, cbdata);

+ ev_break(ctx->event_loop, EVBREAK_ALL);

+rspamd_fuzzy_reply_io(EV_P_ ev_io *w, int revents)

+ struct fuzzy_session *session = (struct fuzzy_session *) w->data;

+ ev_io_stop(EV_A_ w);

+ rspamd_fuzzy_write_reply(session);

+ REF_RELEASE(session);

+rspamd_fuzzy_write_reply(struct fuzzy_session *session)

+ session->cmd_type == CMD_ENCRYPTED_SHINGLE) {

+ len = sizeof(session->reply);

+ len = sizeof(session->reply.hdr) + sizeof(session->reply.rep.v1);

+ len = sizeof(session->reply.rep);

+ len = sizeof(session->reply.rep.v1);

+ r = rspamd_inet_address_sendto(session->fd, data, len, 0,

+ session->addr);

+ REF_RETAIN(session);

+ ev_io_init(&session->io,

+ rspamd_fuzzy_reply_io, session->fd, EV_WRITE);

+ ev_io_start(session->ctx->event_loop, &session->io);

+ msg_err("error while writing reply: %s", strerror(errno));

+ key_stat->errors++;

+ key_stat->matched++;

+ rspamd_set_counter_ema(&key_stat->checked_ctr, nchecked, 0.5f);

+ rspamd_set_counter_ema(&key_stat->matched_ctr, nmatched, 0.5f);

+rspamd_fuzzy_update_stats(struct rspamd_fuzzy_storage_ctx *ctx,

+ enum rspamd_fuzzy_epoch epoch,

+ gboolean matched,

+ gboolean is_shingle,

+ gboolean is_delayed,

+ struct fuzzy_key *key,

+ struct fuzzy_key_stat *ip_stat,

+ guint cmd,

+ struct rspamd_fuzzy_reply *res,

+ ev_tstamp timestamp)

+ ctx->stat.fuzzy_hashes_checked[epoch]++;

+ ctx->stat.delayed_hashes++;

+rspamd_fuzzy_make_reply(struct rspamd_fuzzy_cmd *cmd,

+ struct rspamd_fuzzy_reply *result,

+ struct fuzzy_session *session,

+ gint flags)

+ memcpy(&session->reply.rep, result, sizeof(*result));

+ if (k != kh_end(session->key->forbidden_ids)) {

+ ottery_rand_bytes(session->reply.hdr.nonce,

+ sizeof(session->reply.hdr.nonce));

+ len = sizeof(session->reply.rep);

+ len = sizeof(session->reply.rep.v1);

+ session->epoch,

+ session->reply.rep.v1.prob > 0.5f,

+ flags & RSPAMD_FUZZY_REPLY_SHINGLE,

+ flags & RSPAMD_FUZZY_REPLY_DELAY,

+ session->key,

+ session->ip_stat,

+ cmd->cmd,

+ &session->reply.rep,

+ session->timestamp);

+ }

+

+ rspamd_cryptobox_encrypt_nm_inplace((guchar *) &session->reply.rep,

+ len,

+ session->reply.hdr.nonce,

+ session->nm,

+ session->reply.hdr.mac,

+ RSPAMD_CRYPTOBOX_MODE_25519);

+ session->epoch,

+ session->reply.rep.v1.prob > 0.5f,

+ flags & RSPAMD_FUZZY_REPLY_SHINGLE,

+ flags & RSPAMD_FUZZY_REPLY_DELAY,

+ session->key,

+ session->ip_stat,

+ cmd->cmd,

+ &session->reply.rep,

+ session->timestamp);

+ rspamd_fuzzy_write_reply(session);

+fuzzy_peer_try_send(gint fd, struct fuzzy_peer_request *up_req)

+ r = write(fd, &up_req->cmd, sizeof(up_req->cmd));

+ if (r != sizeof(up_req->cmd)) {

+fuzzy_peer_send_io(EV_P_ ev_io *w, int revents)

+ struct fuzzy_peer_request *up_req = (struct fuzzy_peer_request *) w->data;

+ if (!fuzzy_peer_try_send(w->fd, up_req)) {

+ msg_err("cannot send update request to the peer: %s", strerror(errno));

+ ev_io_stop(EV_A_ w);

+ g_free(up_req);

+rspamd_fuzzy_extensions_tolua(lua_State *L,

+ struct fuzzy_session *session)

+ lua_createtable(L, 0, 0);

+ LL_FOREACH(session->extensions, ext)

+ {

+ lua_pushlstring(L, ext->payload, ext->length);

+ lua_setfield(L, -2, "domain");

+ addr = rspamd_inet_address_new(AF_INET, ext->payload);

+ rspamd_lua_ip_push(L, addr);

+ rspamd_inet_address_free(addr);

+ lua_setfield(L, -2, "ip");

+ addr = rspamd_inet_address_new(AF_INET6, ext->payload);

+ rspamd_lua_ip_push(L, addr);

+ rspamd_inet_address_free(addr);

+ lua_setfield(L, -2, "ip");

+rspamd_fuzzy_check_callback(struct rspamd_fuzzy_reply *result, void *ud)

+ gboolean is_shingle = FALSE, __attribute__((unused)) encrypted = FALSE;

+ memcpy(&sgl_cpy, &session->cmd.sgl, sizeof(sgl_cpy));

+ lua_pushcfunction(L, &rspamd_lua_traceback);

+ err_idx = lua_gettop(L);

+ lua_checkstack(L, err_idx + 9);

+ lua_rawgeti(L, LUA_REGISTRYINDEX, session->ctx->lua_post_handler_cbref);

+ lua_pushnil(L);

+ lua_pushinteger(L, cmd->cmd);

+ (void) lua_new_text(L, result->digest, sizeof(result->digest), FALSE);

+ lua_pushboolean(L, is_shingle);

+ lua_pushinteger(L, result->v1.value);

+ lua_pushnumber(L, result->v1.prob);

+ lua_pushinteger(L, result->v1.flag);

+ lua_pushinteger(L, result->ts);

+ rspamd_fuzzy_extensions_tolua(L, session);

+ if ((ret = lua_pcall(L, 9, LUA_MULTRET, err_idx)) != 0) {

+ msg_err("call to lua_post_handler lua "

+ "script failed (%d): %s",

+ ret, lua_tostring(L, -1));

+ ret = lua_toboolean(L, err_idx + 1);

+ result->v1.value = lua_tointeger(L, err_idx + 2);

+ if (lua_isnumber(L, err_idx + 3)) {

+ result->v1.prob = lua_tonumber(L, err_idx + 3);

+ if (lua_isnumber(L, err_idx + 4)) {

+ result->v1.flag = lua_tointeger(L, err_idx + 4);

+ lua_settop(L, 0);

+ rspamd_fuzzy_make_reply(cmd, result, session, send_flags);

+ REF_RELEASE(session);

+ lua_settop(L, 0);

+ if (!isnan(session->ctx->delay) &&

+ rspamd_match_radix_map_addr(session->ctx->delay_whitelist,

+ session->addr) == NULL) {

+ gdouble hash_age = rspamd_get_calendar_ticks() - result->ts;

+ gdouble jittered_age = rspamd_time_jitter(session->ctx->delay,

+ session->ctx->delay / 2.0);

+ memset(&up_cmd, 0, sizeof(up_cmd));

+ memcpy(up_cmd.cmd.normal.digest, result->digest,

+ sizeof(up_cmd.cmd.normal.digest));

+ memcpy(&up_cmd.cmd.shingle.sgl, shingle,

+ sizeof(up_cmd.cmd.shingle.sgl));

+ g_array_append_val(session->ctx->updates_pending, up_cmd);

+ up_req = g_malloc0(sizeof(*up_req));

+ memcpy(up_req->cmd.cmd.normal.digest, result->digest,

+ sizeof(up_req->cmd.cmd.normal.digest));

+ memcpy(&up_req->cmd.cmd.shingle.sgl, shingle,

+ sizeof(up_req->cmd.cmd.shingle.sgl));

+ if (!fuzzy_peer_try_send(session->ctx->peer_fd, up_req)) {

+ ev_io_init(&up_req->io_ev, fuzzy_peer_send_io,

+ session->ctx->peer_fd, EV_WRITE);

+ ev_io_start(session->ctx->event_loop, &up_req->io_ev);

+ g_free(up_req);

+ rspamd_fuzzy_make_reply(cmd, result, session, send_flags);

+ REF_RELEASE(session);

+rspamd_fuzzy_process_command(struct fuzzy_session *session)

+ gboolean is_shingle = FALSE, __attribute__((unused)) encrypted = FALSE;

+ up_len = sizeof(session->cmd.basic);

+ up_len = sizeof(session->cmd);

+ up_len = sizeof(session->cmd.basic);

+ up_len = sizeof(session->cmd);

+ send_flags |= RSPAMD_FUZZY_REPLY_SHINGLE | RSPAMD_FUZZY_REPLY_ENCRYPTED;

+ msg_err("invalid command type: %d", session->cmd_type);

+ memset(&result, 0, sizeof(result));

+ memcpy(result.digest, cmd->digest, sizeof(result.digest));

+ lua_pushcfunction(L, &rspamd_lua_traceback);

+ err_idx = lua_gettop(L);

+ lua_checkstack(L, err_idx + 5);

+ lua_rawgeti(L, LUA_REGISTRYINDEX, session->ctx->lua_pre_handler_cbref);

+ rspamd_lua_ip_push(L, session->addr);

+ lua_pushinteger(L, cmd->cmd);

+ (void) lua_new_text(L, cmd->digest, sizeof(cmd->digest), FALSE);

+ lua_pushboolean(L, is_shingle);

+ rspamd_fuzzy_extensions_tolua(L, session);

+ if ((ret = lua_pcall(L, 5, LUA_MULTRET, err_idx)) != 0) {

+ msg_err("call to lua_pre_handler lua "

+ "script failed (%d): %s",

+ ret, lua_tostring(L, -1));

+ ret = lua_toboolean(L, err_idx + 1);

+ result.v1.value = lua_tointeger(L, err_idx + 2);

+ if (lua_isnumber(L, err_idx + 3)) {

+ result.v1.prob = lua_tonumber(L, err_idx + 3);

+ lua_settop(L, 0);

+ rspamd_fuzzy_make_reply(cmd, &result, session, send_flags);

+ lua_settop(L, 0);

+ if (G_UNLIKELY(cmd == NULL || up_len == 0)) {

+ rspamd_fuzzy_make_reply(cmd, &result, session, send_flags);

+ rspamd_fuzzy_make_reply(cmd, &result, session, send_flags);

+ ip_stat = rspamd_lru_hash_lookup(session->key->stat->last_ips,

+ session->addr, -1);

+ ip_stat = g_malloc0(sizeof(*ip_stat));

+ REF_INIT_RETAIN(ip_stat, fuzzy_key_stat_dtor);

+ rspamd_lru_hash_insert(session->key->stat->last_ips,

+ naddr, ip_stat, -1, 0);

+ REF_RETAIN(ip_stat);

+ (void) rspamd_fuzzy_check_ratelimit(session); /* Check but ignore */

+ can_continue = rspamd_fuzzy_check_ratelimit(session);

+ REF_RETAIN(session);

+ rspamd_fuzzy_backend_check(session->ctx->backend, cmd,

+ rspamd_fuzzy_check_callback, session);

+ rspamd_fuzzy_make_reply(cmd, &result, session, send_flags);

+ rspamd_fuzzy_make_reply(cmd, &result, session, send_flags);

+ if (rspamd_fuzzy_check_write(session)) {

+ rspamd_encode_hex_buf(cmd->digest, sizeof(cmd->digest),

+ hexbuf, sizeof(hexbuf) - 1);

+ hexbuf[sizeof(hexbuf) - 1] = '\0';

+ if (rspamd_match_hash_map(session->ctx->skip_hashes,

+ hexbuf, sizeof(hexbuf) - 1)) {

+ ptr = is_shingle ? (gpointer) &up_cmd.cmd.shingle : (gpointer) &up_cmd.cmd.normal;

+ memcpy(ptr, cmd, up_len);

+ g_array_append_val(session->ctx->updates_pending, up_cmd);

+ up_req = g_malloc0(sizeof(*up_req));

+ ptr = is_shingle ? (gpointer) &up_req->cmd.cmd.shingle : (gpointer) &up_req->cmd.cmd.normal;

+ memcpy(ptr, cmd, up_len);

+ if (!fuzzy_peer_try_send(session->ctx->peer_fd, up_req)) {

+ ev_io_init(&up_req->io_ev, fuzzy_peer_send_io,

+ session->ctx->peer_fd, EV_WRITE);

+ ev_io_start(session->ctx->event_loop, &up_req->io_ev);

+ g_free(up_req);

+ reply:

+ rspamd_fuzzy_make_reply(cmd, &result, session, send_flags);

+rspamd_fuzzy_command_valid(struct rspamd_fuzzy_cmd *cmd, gint r)

+ if (r >= sizeof(struct rspamd_fuzzy_shingle_cmd)) {

+ if (r >= sizeof(*cmd)) {

+ if (r == sizeof(struct rspamd_fuzzy_shingle_cmd)) {

+ if (r == sizeof(*cmd)) {

+rspamd_fuzzy_decrypt_command(struct fuzzy_session *s, guchar *buf, gsize buflen)

+ msg_warn("received encrypted request when encryption is not enabled");

+ if (buflen < sizeof(hdr)) {

+ msg_warn("XXX: should not be reached");

+ memcpy(&hdr, buf, sizeof(hdr));

+ buf += sizeof(hdr);

+ buflen -= sizeof(hdr);

+ key = g_hash_table_lookup(s->ctx->keys, hdr.key_id);

+ rk = rspamd_pubkey_from_bin(hdr.pubkey, sizeof(hdr.pubkey),

+ RSPAMD_KEYPAIR_KEX, RSPAMD_CRYPTOBOX_MODE_25519);

+ msg_err("bad key; ip=%s",

+ rspamd_inet_address_to_string(s->addr));

+ rspamd_keypair_cache_process(s->ctx->keypair_cache, key->key, rk);

+ if (!rspamd_cryptobox_decrypt_nm_inplace(buf, buflen, hdr.nonce,

+ rspamd_pubkey_get_nm(rk, key->key),

+ hdr.mac, RSPAMD_CRYPTOBOX_MODE_25519)) {

+ msg_err("decryption failed; ip=%s",

+ rspamd_inet_address_to_string(s->addr));

+ rspamd_pubkey_unref(rk);

+ memcpy(s->nm, rspamd_pubkey_get_nm(rk, key->key), sizeof(s->nm));

+ rspamd_pubkey_unref(rk);

+rspamd_fuzzy_extensions_from_wire(struct fuzzy_session *s, guchar *buf, gsize buflen)

+ n_ext++;

+ if (end - p >= sizeof(in_addr_t)) {

+ n_ext++;

+ st_len += sizeof(in_addr_t);

+ p += sizeof(in_addr_t);

+ if (end - p >= sizeof(struct in6_addr)) {

+ n_ext++;

+ st_len += sizeof(struct in6_addr);

+ p += sizeof(struct in6_addr);

+ storage = g_malloc(n_ext * sizeof(struct rspamd_fuzzy_cmd_extension) +

+ st_len);

+ n_ext * sizeof(struct rspamd_fuzzy_cmd_extension);

+ ext = (struct rspamd_fuzzy_cmd_extension *) storage;

+ memcpy(dest, p, dom_len);

+ ext->length = sizeof(in_addr_t);

+ memcpy(dest, p, sizeof(in_addr_t));

+ p += sizeof(in_addr_t);

+ data_buf += sizeof(in_addr_t);

+ ext->length = sizeof(struct in6_addr);

+ memcpy(dest, p, sizeof(struct in6_addr));

+ p += sizeof(struct in6_addr);

+ data_buf += sizeof(struct in6_addr);

+ g_assert_not_reached();

+ ext = (struct rspamd_fuzzy_cmd_extension *) storage;

+rspamd_fuzzy_cmd_from_wire(guchar *buf, guint buflen, struct fuzzy_session *s)

+ if (buflen < sizeof(struct rspamd_fuzzy_cmd)) {

+ msg_debug("truncated fuzzy command of size %d received", buflen);

+ if (buflen >= sizeof(struct rspamd_fuzzy_encrypted_cmd)) {

+ if (memcmp(buf, fuzzy_encrypted_magic, sizeof(fuzzy_encrypted_magic)) == 0) {

+ if (!rspamd_fuzzy_decrypt_command(s, buf, buflen)) {

+ buf += sizeof(struct rspamd_fuzzy_encrypted_req_hdr);

+ buflen -= sizeof(struct rspamd_fuzzy_encrypted_req_hdr);

+ if (buflen < sizeof(s->cmd.basic)) {

+ msg_debug("truncated normal fuzzy command of size %d received", buflen);

+ memcpy(&s->cmd.basic, buf, sizeof(s->cmd.basic));

+ epoch = rspamd_fuzzy_command_valid(&s->cmd.basic, buflen);

+ msg_debug("invalid fuzzy command of size %d received", buflen);

+ buf += sizeof(s->cmd.basic);

+ buflen -= sizeof(s->cmd.basic);

+ if (buflen >= sizeof(s->cmd.sgl)) {

+ memcpy(&s->cmd.sgl, buf, sizeof(s->cmd.sgl));

+ msg_debug("truncated fuzzy shingles command of size %d received", buflen);

+ buf += sizeof(s->cmd.sgl);

+ buflen -= sizeof(s->cmd.sgl);

+ if (!rspamd_fuzzy_extensions_from_wire(s, buf, buflen)) {

+ msg_debug("truncated fuzzy shingles command of size %d received", buflen);

+fuzzy_session_destroy(gpointer d)

+ rspamd_inet_address_free(session->addr);

+ rspamd_explicit_memzero(session->nm, sizeof(session->nm));

+ REF_RELEASE(session->ip_stat);

+ g_free(session->extensions);

+ g_free(session);

+accept_fuzzy_socket(EV_P_ ev_io *w, int revents)

+ struct rspamd_worker *worker = (struct rspamd_worker *) w->data;

+ socklen_t salen = sizeof(peer_sa[0]);

+ memset(msg, 0, sizeof(*msg) * MSGVEC_LEN);

+ ctx = (struct rspamd_fuzzy_storage_ctx *) worker->ctx;

+ for (int i = 0; i < MSGVEC_LEN; i++) {

+ iovs[i].iov_len = sizeof(bufs[i]);

+ MSG_FIELD(msg[i], msg_name) = (void *) &peer_sa[i];

+ ev_now_update_if_cheap(ctx->event_loop);

+ r = recvmmsg(w->fd, msg, MSGVEC_LEN, 0, NULL);

+ r = recvmsg(w->fd, msg, 0);

+ msg_err("got error while reading from socket: %d, %s",

+ strerror(errno));

+ r = 1; /* Assume that we have received a single message */

+ for (int i = 0; i < r; i++) {

+ MSG_FIELD(msg[i], msg_namelen));

+ if (!rspamd_fuzzy_check_client(worker->ctx, client_addr)) {

+ rspamd_inet_address_free(client_addr);

+ session = g_malloc0(sizeof(*session));

+ REF_INIT_RETAIN(session, fuzzy_session_destroy);

+ session->timestamp = ev_now(ctx->event_loop);

+ if (rspamd_fuzzy_cmd_from_wire(iovs[i].iov_base,

+ msg_len, session)) {

+ rspamd_fuzzy_process_command(session);

+ session->ctx->stat.invalid_requests++;

+ msg_debug("invalid fuzzy command of size %z received", r);

+ session->addr, -1);

+ rspamd_inet_address_copy(session->addr, NULL),

+ nerrors, -1, -1);

+ REF_RELEASE(session);

+rspamd_fuzzy_storage_periodic_callback(void *ud)

+ rspamd_fuzzy_process_updates_queue(ctx, local_db_name, FALSE);

+rspamd_fuzzy_storage_sync(struct rspamd_main *rspamd_main,

+ struct rspamd_worker *worker, gint fd,

+ gint attached_fd,

+ struct rspamd_control_command *cmd,

+ gpointer ud)

+ rspamd_fuzzy_process_updates_queue(ctx, local_db_name, FALSE);

+ rspamd_fuzzy_backend_start_update(ctx->backend, ctx->sync_timeout,

+ rspamd_fuzzy_storage_periodic_callback, ctx);

+ if (write(fd, &rep, sizeof(rep)) != sizeof(rep)) {

+ msg_err("cannot write reply to the control socket: %s",

+ strerror(errno));

+rspamd_fuzzy_control_blocked(struct rspamd_main *rspamd_main,

+ struct rspamd_worker *worker, gint fd,

+ gint attached_fd,

+ struct rspamd_control_command *cmd,

+ gpointer ud)

+ struct rspamd_fuzzy_storage_ctx *ctx = (struct rspamd_fuzzy_storage_ctx *) ud;

+ ev_tstamp now = ev_now(ctx->event_loop);

+ sizeof(struct sockaddr_in));

+ sizeof(struct sockaddr_in6));

+ msg_err("invalid address family: %d", cmd->cmd.fuzzy_blocked.af);

+ (time_t) now);

+ if (isnan(elt->cur)) {

+ msg_info("propagating ratelimiting %s, %.1f max elts",

+ rspamd_inet_address_to_string(addr),

+ ctx->leaky_bucket_burst);

+ addr,

+ elt,

+ (time_t) now,

+ ctx->leaky_bucket_ttl);

+ msg_info("propagating ratelimiting %s, %.1f max elts",

+ rspamd_inet_address_to_string(addr),

+ ctx->leaky_bucket_burst);

+ if (write(fd, &rep, sizeof(rep)) != sizeof(rep)) {

+ msg_err("cannot write reply to the control socket: %s",

+ strerror(errno));

+rspamd_fuzzy_storage_reload(struct rspamd_main *rspamd_main,

+ struct rspamd_worker *worker, gint fd,

+ gint attached_fd,

+ struct rspamd_control_command *cmd,

+ gpointer ud)

+ msg_info("reloading fuzzy storage after receiving reload command");

+ rspamd_fuzzy_backend_close(ctx->backend);

+ memset(&rep, 0, sizeof(rep));

+ if ((ctx->backend = rspamd_fuzzy_backend_create(ctx->event_loop,

+ worker->cf->options, rspamd_main->cfg,

+ &err)) == NULL) {

+ msg_err("cannot open backend after reload: %e", err);

+ g_error_free(err);

+ rspamd_fuzzy_backend_start_update(ctx->backend, ctx->sync_timeout,

+ rspamd_fuzzy_storage_periodic_callback, ctx);

+ if (write(fd, &rep, sizeof(rep)) != sizeof(rep)) {

+ msg_err("cannot write reply to the control socket: %s",

+ strerror(errno));

+rspamd_fuzzy_storage_stat_key(const struct fuzzy_key_stat *key_stat)

+ res = ucl_object_typed_new(UCL_OBJECT);

+

+ ucl_object_insert_key(res, ucl_object_fromint(key_stat->checked),

+ "checked", 0, false);

+ ucl_object_insert_key(res, ucl_object_fromdouble(key_stat->checked_ctr.mean),

+ "checked_per_hour", 0, false);

+ ucl_object_insert_key(res, ucl_object_fromint(key_stat->matched),

+ "matched", 0, false);

+ ucl_object_insert_key(res, ucl_object_fromdouble(key_stat->matched_ctr.mean),

+ "matched_per_hour", 0, false);

+ ucl_object_insert_key(res, ucl_object_fromint(key_stat->added),

+ "added", 0, false);

+ ucl_object_insert_key(res, ucl_object_fromint(key_stat->deleted),

+ "deleted", 0, false);

+ ucl_object_insert_key(res, ucl_object_fromint(key_stat->errors),

+ "errors", 0, false);

+rspamd_fuzzy_stat_to_ucl(struct rspamd_fuzzy_storage_ctx *ctx, gboolean ip_stat)

+ obj = ucl_object_typed_new(UCL_OBJECT);

+ keys_obj = ucl_object_typed_new(UCL_OBJECT);

+ g_hash_table_iter_init(&it, ctx->keys);

+ while (g_hash_table_iter_next(&it, &k, &v)) {

+ rspamd_snprintf(keyname, sizeof(keyname), "%8bs", k);

+ elt = rspamd_fuzzy_storage_stat_key(key_stat);

+ ip_elt = ucl_object_typed_new(UCL_OBJECT);

+ while ((i = rspamd_lru_hash_foreach(key_stat->last_ips,

+ i, &k, &v)) != -1) {

+ ip_cur = rspamd_fuzzy_storage_stat_key(v);

+ ucl_object_insert_key(ip_elt, ip_cur,

+ rspamd_inet_address_to_string(k), 0, true);

+ ucl_object_insert_key(elt, ip_elt, "ips", 0, false);

+ ucl_object_insert_key(flags_ucl, rspamd_fuzzy_storage_stat_key(flag_stat),

+ intbuf, 0, true);

+ ucl_object_insert_key(elt, flags_ucl, "flags", 0, false);

+ ucl_object_insert_key(elt,

+ rspamd_keypair_to_ucl(fuzzy_key->key, RSPAMD_KEYPAIR_DUMP_NO_SECRET | RSPAMD_KEYPAIR_DUMP_FLATTENED),

+ "keypair", 0, false);

+ ucl_object_insert_key(keys_obj, elt, keyname, 0, true);

+ ucl_object_insert_key(obj, keys_obj, "keys", 0, false);

+ ucl_object_insert_key(obj,

+ ucl_object_fromint(ctx->stat.fuzzy_hashes),

+ "fuzzy_stored",

+ 0,

+ false);

+ ucl_object_insert_key(obj,

+ ucl_object_fromint(ctx->stat.fuzzy_hashes_expired),

+ "fuzzy_expired",

+ 0,

+ false);

+ ucl_object_insert_key(obj,

+ ucl_object_fromint(ctx->stat.invalid_requests),

+ "invalid_requests",

+ 0,

+ false);

+ ucl_object_insert_key(obj,

+ ucl_object_fromint(ctx->stat.delayed_hashes),

+ "delayed_hashes",

+ 0,

+ false);

+ ip_elt = ucl_object_typed_new(UCL_OBJECT);

+ while ((i = rspamd_lru_hash_foreach(ctx->errors_ips, i, &k, &v)) != -1) {

+ ucl_object_insert_key(ip_elt,

+ ucl_object_fromint(*(guint64 *) v),

+ rspamd_inet_address_to_string(k), 0, true);

+ ucl_object_insert_key(obj,

+ ip_elt,

+ "errors_ips",

+ 0,

+ false);

+ elt = ucl_object_typed_new(UCL_ARRAY);

+ ucl_array_append(elt,

+ ucl_object_fromint(ctx->stat.fuzzy_hashes_checked[i]));

+ ucl_object_insert_key(obj, elt, "fuzzy_checked", 0, false);

+ elt = ucl_object_typed_new(UCL_ARRAY);

+ ucl_array_append(elt,

+ ucl_object_fromint(ctx->stat.fuzzy_shingles_checked[i]));

+ ucl_object_insert_key(obj, elt, "fuzzy_shingles", 0, false);

+ elt = ucl_object_typed_new(UCL_ARRAY);

+ ucl_array_append(elt,

+ ucl_object_fromint(ctx->stat.fuzzy_hashes_found[i]));

+ ucl_object_insert_key(obj, elt, "fuzzy_found", 0, false);

+rspamd_fuzzy_maybe_load_ratelimits(struct rspamd_fuzzy_storage_ctx *ctx)

+ RSPAMD_DBDIR);

+ if (access(path, R_OK) != -1) {

+ struct ucl_parser *parser = ucl_parser_new(UCL_PARSER_NO_IMPLICIT_ARRAYS | UCL_PARSER_DISABLE_MACRO);

+ RSPAMD_INET_ADDRESS_PARSE_NO_UNIX | RSPAMD_INET_ADDRESS_PARSE_NO_PORT)) {

+ loaded++;

+rspamd_fuzzy_maybe_save_ratelimits(struct rspamd_fuzzy_storage_ctx *ctx)

+ RSPAMD_DBDIR);

+ struct rspamd_leaky_bucket_elt *elt = (struct rspamd_leaky_bucket_elt *) v;

+ RSPAMD_DBDIR);

+lua_fuzzy_add_pre_handler(lua_State *L)

+ rspamd_lua_check_udata(L, 1, "rspamd{worker}");

+ return luaL_error(L, "invalid arguments, worker + function are expected");

+ if (wrk && lua_isfunction(L, 2)) {

+ ctx = (struct rspamd_fuzzy_storage_ctx *) wrk->ctx;

+ luaL_unref(L, LUA_REGISTRYINDEX, ctx->lua_pre_handler_cbref);

+ lua_pushvalue(L, 2);

+ ctx->lua_pre_handler_cbref = luaL_ref(L, LUA_REGISTRYINDEX);

+ return luaL_error(L, "invalid arguments, worker + function are expected");

+lua_fuzzy_add_post_handler(lua_State *L)

+ rspamd_lua_check_udata(L, 1, "rspamd{worker}");

+ return luaL_error(L, "invalid arguments, worker + function are expected");

+ if (wrk && lua_isfunction(L, 2)) {

+ ctx = (struct rspamd_fuzzy_storage_ctx *) wrk->ctx;

+ luaL_unref(L, LUA_REGISTRYINDEX, ctx->lua_post_handler_cbref);

+ lua_pushvalue(L, 2);

+ ctx->lua_post_handler_cbref = luaL_ref(L, LUA_REGISTRYINDEX);

+ return luaL_error(L, "invalid arguments, worker + function are expected");

+lua_fuzzy_add_blacklist_handler(lua_State *L)

+ rspamd_lua_check_udata(L, 1, "rspamd{worker}");

+ return luaL_error(L, "invalid arguments, worker + function are expected");

+ if (wrk && lua_isfunction(L, 2)) {

+ ctx = (struct rspamd_fuzzy_storage_ctx *) wrk->ctx;

+ luaL_unref(L, LUA_REGISTRYINDEX, ctx->lua_blacklist_cbref);

+ lua_pushvalue(L, 2);

+ ctx->lua_blacklist_cbref = luaL_ref(L, LUA_REGISTRYINDEX);

+ return luaL_error(L, "invalid arguments, worker + function are expected");

+rspamd_fuzzy_storage_stat(struct rspamd_main *rspamd_main,

+ struct rspamd_worker *worker, gint fd,

+ gint attached_fd,

+ struct rspamd_control_command *cmd,

+ gpointer ud)

+ guchar fdspace[CMSG_SPACE(sizeof(int))];

+ memset(&rep, 0, sizeof(rep));

+ rspamd_snprintf(tmppath, sizeof(tmppath), "%s%c%s-XXXXXXXXXX",

+ rspamd_main->cfg->temp_dir, G_DIR_SEPARATOR, "fuzzy-stat");

+ if ((outfd = mkstemp(tmppath)) == -1) {

+ msg_info_main("cannot make temporary stat file for fuzzy stat: %s",

+ strerror(errno));

+ memcpy(rep.reply.fuzzy_stat.storage_id,

+ rspamd_fuzzy_backend_id(ctx->backend),

+ sizeof(rep.reply.fuzzy_stat.storage_id));

+ obj = rspamd_fuzzy_stat_to_ucl(ctx, TRUE);

+ emit_subr = ucl_object_emit_fd_funcs(outfd);

+ ucl_object_emit_full(obj, UCL_EMIT_JSON_COMPACT, emit_subr, NULL);

+ ucl_object_emit_funcs_free(emit_subr);

+ ucl_object_unref(obj);

+ close(outfd);

+ outfd = open(tmppath, O_RDONLY);

+ unlink(tmppath);

+ memset(&msg, 0, sizeof(msg));

+ memset(fdspace, 0, sizeof(fdspace));

+ msg.msg_controllen = sizeof(fdspace);

+ cmsg = CMSG_FIRSTHDR(&msg);

+ cmsg->cmsg_len = CMSG_LEN(sizeof(int));

+ memcpy(CMSG_DATA(cmsg), &outfd, sizeof(int));

+ iov.iov_len = sizeof(rep);

+ if (sendmsg(fd, &msg, 0) == -1) {

+ msg_err_main("cannot send fuzzy stat: %s", strerror(errno));

+ close(outfd);

+fuzzy_parse_ids(rspamd_mempool_t *pool,

+ const ucl_object_t *obj,

+ gpointer ud,

+ struct rspamd_rcl_section *section,

+ GError **err)

+ struct rspamd_rcl_struct_parser *pd = (struct rspamd_rcl_struct_parser *) ud;

+ khash_t(fuzzy_key_ids_set) * target;

+ target = *(khash_t(fuzzy_key_ids_set) **) ((gchar *) pd->user_struct + pd->offset);

+ if (ucl_object_type(obj) == UCL_ARRAY) {

+ while ((cur = ucl_object_iterate(obj, &it, true)) != NULL) {

+ if (ucl_object_toint_safe(cur, &id)) {

+ else if (ucl_object_type(obj) == UCL_INT) {

+ kh_put(fuzzy_key_ids_set, target, ucl_object_toint(obj), &r);

+fuzzy_parse_keypair(rspamd_mempool_t *pool,

+ const ucl_object_t *obj,

+ gpointer ud,

+ struct rspamd_rcl_section *section,

+ GError **err)

+ pd->offset = G_STRUCT_OFFSET(struct rspamd_fuzzy_storage_ctx, default_keypair);

+ if (ucl_object_type(obj) == UCL_STRING || ucl_object_type(obj) == UCL_OBJECT) {

+ ret = rspamd_rcl_parse_struct_keypair(pool, obj, pd, section, err);

+ if (rspamd_keypair_alg(kp) != RSPAMD_CRYPTOBOX_MODE_25519 ||

+ rspamd_keypair_type(kp) != RSPAMD_KEYPAIR_KEX) {

+ key = g_malloc0(sizeof(*key));

+ keystat = g_malloc0(sizeof(*keystat));

+ REF_INIT_RETAIN(keystat, fuzzy_key_stat_dtor);

+ keystat->last_ips = rspamd_lru_hash_new_full(1024,

+ (GDestroyNotify) rspamd_inet_address_free,

+ fuzzy_key_stat_unref,

+ rspamd_inet_address_hash, rspamd_inet_address_equal);

+ pk = rspamd_keypair_component(kp, RSPAMD_KEYPAIR_COMPONENT_PK,

+ NULL);

+ g_hash_table_insert(ctx->keys, (gpointer) pk, key);

+ const ucl_object_t *forbidden_ids = ucl_object_lookup(extensions, "forbidden_ids");

+ if (forbidden_ids && ucl_object_type(forbidden_ids) == UCL_ARRAY) {

+ while ((cur = ucl_object_iterate(forbidden_ids, &it, true)) != NULL) {

+ else if (ucl_object_type(obj) == UCL_ARRAY) {

+ while ((cur = ucl_object_iterate(obj, &it, true)) != NULL) {

+ if (!fuzzy_parse_keypair(pool, cur, pd, section, err)) {

+ msg_err_pool("cannot parse keypair");

+fuzzy_kp_hash(gconstpointer p)

+ return *(guint *) p;

+fuzzy_kp_equal(gconstpointer a, gconstpointer b)

+ return (memcmp(pa, pb, RSPAMD_FUZZY_KEYLEN) == 0);

+init_fuzzy(struct rspamd_config *cfg)

+ type = g_quark_try_string("fuzzy");

+ ctx = rspamd_mempool_alloc0(cfg->cfg_pool,

+ sizeof(struct rspamd_fuzzy_storage_ctx));

+ ctx->keys = g_hash_table_new_full(fuzzy_kp_hash, fuzzy_kp_equal,

+ NULL, fuzzy_key_dtor);

+ rspamd_mempool_add_destructor(cfg->cfg_pool,

+ (rspamd_mempool_destruct_t) g_hash_table_unref, ctx->keys);

+ ctx->errors_ips = rspamd_lru_hash_new_full(1024,

+ (GDestroyNotify) rspamd_inet_address_free, g_free,

+ rspamd_inet_address_hash, rspamd_inet_address_equal);

+ rspamd_mempool_add_destructor(cfg->cfg_pool,

+ (rspamd_mempool_destruct_t) rspamd_lru_hash_destroy, ctx->errors_ips);

+ rspamd_rcl_register_worker_option(cfg,

+ type,

+ "sync",

+ rspamd_rcl_parse_struct_time,

+ ctx,

+ G_STRUCT_OFFSET(struct rspamd_fuzzy_storage_ctx,

+ sync_timeout),

+ RSPAMD_CL_FLAG_TIME_FLOAT,

+ "Time to perform database sync, default: " G_STRINGIFY(DEFAULT_SYNC_TIMEOUT) " seconds");

+

+ rspamd_rcl_register_worker_option(cfg,

+ type,

+ "expire",

+ rspamd_rcl_parse_struct_time,

+ ctx,

+ G_STRUCT_OFFSET(struct rspamd_fuzzy_storage_ctx,

+ expire),

+ RSPAMD_CL_FLAG_TIME_FLOAT,

+ "Default expire time for hashes, default: " G_STRINGIFY(DEFAULT_EXPIRE) " seconds");

+

+ rspamd_rcl_register_worker_option(cfg,

+ type,

+ "delay",

+ rspamd_rcl_parse_struct_time,

+ ctx,

+ G_STRUCT_OFFSET(struct rspamd_fuzzy_storage_ctx,

+ delay),

+ RSPAMD_CL_FLAG_TIME_FLOAT,

+ "Default delay time for hashes, default: not enabled");

+

+ rspamd_rcl_register_worker_option(cfg,

+ type,

+ "allow_update",

+ rspamd_rcl_parse_struct_ucl,

+ ctx,

+ G_STRUCT_OFFSET(struct rspamd_fuzzy_storage_ctx, update_map),

+ 0,

+ "Allow modifications from the following IP addresses");

+

+ rspamd_rcl_register_worker_option(cfg,

+ type,

+ "allow_update_keys",

+ rspamd_rcl_parse_struct_ucl,

+ ctx,

+ G_STRUCT_OFFSET(struct rspamd_fuzzy_storage_ctx, update_keys_map),

+ 0,

+ "Allow modifications for those using specific public keys");

+

+ rspamd_rcl_register_worker_option(cfg,

+ type,

+ "delay_whitelist",

+ rspamd_rcl_parse_struct_ucl,

+ ctx,

+ G_STRUCT_OFFSET(struct rspamd_fuzzy_storage_ctx, delay_whitelist_map),

+ 0,

+ "Disable delay check for the following IP addresses");

+

+ rspamd_rcl_register_worker_option(cfg,

+ type,

+ "keypair",

+ fuzzy_parse_keypair,

+ ctx,

+ 0,

+ RSPAMD_CL_FLAG_MULTIPLE,

+ "Encryption keypair (can be repeated for different keys)");

+

+ rspamd_rcl_register_worker_option(cfg,

+ type,

+ "forbidden_ids",

+ fuzzy_parse_ids,

+ ctx,

+ G_STRUCT_OFFSET(struct rspamd_fuzzy_storage_ctx, default_forbidden_ids),

+ 0,

+ "Deny specific flags by default");

+

+ rspamd_rcl_register_worker_option(cfg,

+ type,

+ "weak_ids",

+ fuzzy_parse_ids,

+ ctx,

+ G_STRUCT_OFFSET(struct rspamd_fuzzy_storage_ctx, weak_ids),

+ 0,

+ "Treat these flags as weak (i.e. they do not overwrite strong flags)");

+

+ rspamd_rcl_register_worker_option(cfg,

+ type,

+ "keypair_cache_size",

+ rspamd_rcl_parse_struct_integer,

+ ctx,

+ G_STRUCT_OFFSET(struct rspamd_fuzzy_storage_ctx,

+ keypair_cache_size),

+ RSPAMD_CL_FLAG_UINT,

+ "Size of keypairs cache, default: " G_STRINGIFY(DEFAULT_KEYPAIR_CACHE_SIZE));

+

+ rspamd_rcl_register_worker_option(cfg,

+ type,

+ "encrypted_only",

+ rspamd_rcl_parse_struct_boolean,

+ ctx,

+ G_STRUCT_OFFSET(struct rspamd_fuzzy_storage_ctx, encrypted_only),

+ 0,

+ "Allow encrypted requests only (and forbid all unknown keys or plaintext requests)");

+ rspamd_rcl_register_worker_option(cfg,

+ type,

+ "dedicated_update_worker",

+ rspamd_rcl_parse_struct_boolean,

+ ctx,

+ G_STRUCT_OFFSET(struct rspamd_fuzzy_storage_ctx, dedicated_update_worker),

+ 0,

+ "Use worker 0 for updates only");

+

+ rspamd_rcl_register_worker_option(cfg,

+ type,

+ "read_only",

+ rspamd_rcl_parse_struct_boolean,

+ ctx,

+ G_STRUCT_OFFSET(struct rspamd_fuzzy_storage_ctx, read_only),

+ 0,

+ "Work in read only mode");

+

+

+ rspamd_rcl_register_worker_option(cfg,

+ type,

+ "blocked",

+ rspamd_rcl_parse_struct_ucl,

+ ctx,

+ G_STRUCT_OFFSET(struct rspamd_fuzzy_storage_ctx, blocked_map),

+ 0,

+ "Block requests from specific networks");

+

+

+ rspamd_rcl_register_worker_option(cfg,

+ type,

+ "updates_maxfail",

+ rspamd_rcl_parse_struct_integer,

+ ctx,

+ G_STRUCT_OFFSET(struct rspamd_fuzzy_storage_ctx, updates_maxfail),

+ RSPAMD_CL_FLAG_UINT,

+ "Maximum number of updates to be failed before discarding");

+ rspamd_rcl_register_worker_option(cfg,

+ type,

+ "skip_hashes",

+ rspamd_rcl_parse_struct_ucl,

+ ctx,

+ G_STRUCT_OFFSET(struct rspamd_fuzzy_storage_ctx, skip_map),

+ 0,

+ "Skip specific hashes from the map");

+ rspamd_rcl_register_worker_option(cfg,

+ type,

+ "ratelimit_whitelist",

+ rspamd_rcl_parse_struct_ucl,

+ ctx,

+ G_STRUCT_OFFSET(struct rspamd_fuzzy_storage_ctx, ratelimit_whitelist_map),

+ 0,

+ "Skip specific addresses from rate limiting");

+ rspamd_rcl_register_worker_option(cfg,

+ type,

+ "ratelimit_max_buckets",

+ rspamd_rcl_parse_struct_integer,

+ ctx,

+ G_STRUCT_OFFSET(struct rspamd_fuzzy_storage_ctx, max_buckets),

+ RSPAMD_CL_FLAG_UINT,

+ "Maximum number of leaky buckets (default: " G_STRINGIFY(DEFAULT_MAX_BUCKETS) ")");

+ rspamd_rcl_register_worker_option(cfg,

+ type,

+ "ratelimit_network_mask",

+ rspamd_rcl_parse_struct_integer,

+ ctx,

+ G_STRUCT_OFFSET(struct rspamd_fuzzy_storage_ctx, leaky_bucket_mask),

+ RSPAMD_CL_FLAG_UINT,

+ "Network mask to apply for IPv4 rate addresses (default: " G_STRINGIFY(DEFAULT_BUCKET_MASK) ")");

+ rspamd_rcl_register_worker_option(cfg,

+ type,

+ "ratelimit_bucket_ttl",

+ rspamd_rcl_parse_struct_time,

+ ctx,

+ G_STRUCT_OFFSET(struct rspamd_fuzzy_storage_ctx, leaky_bucket_ttl),

+ RSPAMD_CL_FLAG_TIME_INTEGER,

+ "Time to live for ratelimit element (default: " G_STRINGIFY(DEFAULT_BUCKET_TTL) ")");

+ rspamd_rcl_register_worker_option(cfg,

+ type,

+ "ratelimit_rate",

+ rspamd_rcl_parse_struct_double,

+ ctx,

+ G_STRUCT_OFFSET(struct rspamd_fuzzy_storage_ctx, leaky_bucket_rate),

+ 0,

+ "Leak rate in requests per second");

+ rspamd_rcl_register_worker_option(cfg,

+ type,

+ "ratelimit_burst",

+ rspamd_rcl_parse_struct_double,

+ ctx,

+ G_STRUCT_OFFSET(struct rspamd_fuzzy_storage_ctx, leaky_bucket_burst),

+ 0,

+ "Peak value for ratelimit bucket");

+ rspamd_rcl_register_worker_option(cfg,

+ type,

+ "ratelimit_log_only",

+ rspamd_rcl_parse_struct_boolean,

+ ctx,

+ G_STRUCT_OFFSET(struct rspamd_fuzzy_storage_ctx, ratelimit_log_only),

+ 0,

+ "Don't really ban on ratelimit reaching, just log");

+rspamd_fuzzy_peer_io(EV_P_ ev_io *w, int revents)

+ (struct rspamd_fuzzy_storage_ctx *) w->data;

+ r = read(w->fd, &cmd, sizeof(cmd));

+ if (r != sizeof(cmd)) {

+ msg_err("cannot read command from peers: %s", strerror(errno));

+ g_array_append_val(ctx->updates_pending, cmd);

+fuzzy_peer_rep(struct rspamd_worker *worker,

+ struct rspamd_srv_reply *rep, gint rep_fd,

+ gpointer ud)

+ msg_err("cannot receive peer fd from the main process");

+ exit(EXIT_FAILURE);

+ rspamd_socket_nonblocking(rep_fd);

+ msg_info("got peer fd reply from the main process");

+ msg_info("start listening on %s",

+ rspamd_inet_address_to_string_pretty(ls->addr));

+ ac_ev = g_malloc0(sizeof(*ac_ev));

+ ev_io_init(&ac_ev->accept_ev, accept_fuzzy_socket, ls->fd,

+ EV_READ);

+ ev_io_start(ctx->event_loop, &ac_ev->accept_ev);

+ DL_APPEND(worker->accept_events, ac_ev);

+ else {

+ g_assert_not_reached();

+ cur = g_list_next(cur);

+ shutdown(ctx->peer_fd, SHUT_WR);

+ ev_io_init(&ctx->peer_ev, rspamd_fuzzy_peer_io, ctx->peer_fd, EV_READ);

+ ev_io_start(ctx->event_loop, &ctx->peer_ev);

+ shutdown(ctx->peer_fd, SHUT_RD);

+__attribute__((noreturn)) void

+start_fuzzy(struct rspamd_worker *worker)

+ g_assert(rspamd_worker_check_context(worker->ctx, rspamd_fuzzy_storage_magic));

+ ctx->event_loop = rspamd_prepare_worker(worker,

+ "fuzzy",

+ NULL);

+ ctx->resolver = rspamd_dns_resolver_init(worker->srv->logger,

+ ctx->event_loop,

+ worker->srv->cfg);

+ rspamd_upstreams_library_config(worker->srv->cfg, ctx->cfg->ups_ctx,

+ ctx->event_loop, ctx->resolver->r);

+ ctx->http_ctx = rspamd_http_context_create(ctx->cfg, ctx->event_loop,

+ ctx->cfg->ups_ctx);

+ ctx->keypair_cache = rspamd_keypair_cache_new(ctx->keypair_cache_size);

+ if ((ctx->backend = rspamd_fuzzy_backend_create(ctx->event_loop,

+ worker->cf->options, cfg, &err)) == NULL) {

+ msg_err("cannot open backend: %e", err);

+ g_error_free(err);

+ exit(EXIT_SUCCESS);

+ rspamd_fuzzy_backend_count(ctx->backend, fuzzy_count_callback, ctx);

+ ctx->updates_pending = g_array_sized_new(FALSE, FALSE,

+ sizeof(struct fuzzy_peer_cmd), 1024);

+ rspamd_fuzzy_backend_start_update(ctx->backend, ctx->sync_timeout,

+ rspamd_fuzzy_storage_periodic_callback, ctx);

+ msg_info_config("stop serving clients request in dedicated update mode");

+ rspamd_worker_stop_accept(worker);

+ (struct rspamd_worker_listen_socket *) cur->data;

+ close(ls->fd);

+ cur = g_list_next(cur);

+ ev_timer_init(&ctx->stat_ev, rspamd_fuzzy_stat_callback, ctx->sync_timeout,

+ ctx->sync_timeout);

+ ev_timer_start(ctx->event_loop, &ctx->stat_ev);

+ rspamd_control_worker_add_cmd_handler(worker, RSPAMD_CONTROL_RELOAD,

+ rspamd_fuzzy_storage_reload, ctx);

+ rspamd_control_worker_add_cmd_handler(worker, RSPAMD_CONTROL_FUZZY_STAT,

+ rspamd_fuzzy_storage_stat, ctx);

+ rspamd_control_worker_add_cmd_handler(worker, RSPAMD_CONTROL_FUZZY_SYNC,

+ rspamd_fuzzy_storage_sync, ctx);

+ rspamd_control_worker_add_cmd_handler(worker, RSPAMD_CONTROL_FUZZY_BLOCKED,

+ rspamd_fuzzy_control_blocked, ctx);

+ rspamd_config_radix_from_ucl(worker->srv->cfg, ctx->update_map,

+ "Allow fuzzy updates from specified addresses",

+ &ctx->update_ips, NULL, worker, "fuzzy update");

+ if ((m = rspamd_map_add_from_ucl(worker->srv->cfg, ctx->update_keys_map,

+ "Allow fuzzy updates from specified public keys",

+ rspamd_kv_list_read,

+ rspamd_kv_list_fin,

+ rspamd_kv_list_dtor,

+ (void **) &ctx->update_keys, worker, RSPAMD_MAP_DEFAULT)) == NULL) {

+ msg_warn_config("cannot load allow keys map from %s",

+ ucl_object_tostring(ctx->update_keys_map));

+ if ((m = rspamd_map_add_from_ucl(cfg, ctx->skip_map,

+ "Skip hashes",

+ rspamd_kv_list_read,

+ rspamd_kv_list_fin,

+ rspamd_kv_list_dtor,

+ (void **) &ctx->skip_hashes,

+ worker, RSPAMD_MAP_DEFAULT)) == NULL) {

+ msg_warn_config("cannot load hashes list from %s",

+ ucl_object_tostring(ctx->skip_map));

+ rspamd_config_radix_from_ucl(worker->srv->cfg, ctx->blocked_map,

+ "Block fuzzy requests from the specific IPs",

+ &ctx->blocked_ips,

+ NULL,

+ worker, "fuzzy blocked");

+ rspamd_config_radix_from_ucl(worker->srv->cfg, ctx->ratelimit_whitelist_map,

+ "Skip ratelimits from specific ip addresses/networks",

+ &ctx->ratelimit_whitelist,

+ NULL,

+ worker, "fuzzy ratelimit whitelist");

+ if (!isnan(ctx->delay) && ctx->delay_whitelist_map != NULL) {

+ rspamd_config_radix_from_ucl(worker->srv->cfg, ctx->delay_whitelist_map,

+ "Skip delay from the following ips",

+ &ctx->delay_whitelist, NULL, worker,

+ "fuzzy delayed whitelist");

+ if (!isnan(ctx->leaky_bucket_rate) && !isnan(ctx->leaky_bucket_burst)) {

+ ctx->ratelimit_buckets = rspamd_lru_hash_new_full(ctx->max_buckets,

+ NULL, fuzzy_rl_bucket_free,

+ rspamd_inet_address_hash, rspamd_inet_address_equal);

+ rspamd_fuzzy_maybe_load_ratelimits(ctx);

+ ctx->resolver = rspamd_dns_resolver_init(worker->srv->logger,

+ ctx->event_loop,

+ worker->srv->cfg);

+ rspamd_map_watch(worker->srv->cfg, ctx->event_loop,

+ ctx->resolver, worker, RSPAMD_MAP_WATCH_WORKER);

+ memset(&srv_cmd, 0, sizeof(srv_cmd));

+ memset(srv_cmd.cmd.spair.pair_id, 0, sizeof(srv_cmd.cmd.spair.pair_id));

+ G_STATIC_ASSERT(sizeof(srv_cmd.cmd.spair.pair_id) >=

+ sizeof("fuzzy") + sizeof(guint64));

+ memcpy(srv_cmd.cmd.spair.pair_id, "fuzzy", sizeof("fuzzy"));

+ guint64 bind_hash = rspamd_cryptobox_fast_hash(worker->cf->bind_conf->bind_line,

+ strlen(worker->cf->bind_conf->bind_line), 0xdeadbabe);

+ memcpy(srv_cmd.cmd.spair.pair_id + sizeof("fuzzy"), &bind_hash,

+ sizeof(bind_hash));

+ rspamd_srv_send_command(worker, ctx->event_loop, &srv_cmd, -1,

+ fuzzy_peer_rep, ctx);

+ .name = "add_fuzzy_pre_handler",

+ .func = lua_fuzzy_add_pre_handler,

+ rspamd_lua_add_metamethod(ctx->cfg->lua_state, "rspamd{worker}", &fuzzy_lua_reg);

+ .name = "add_fuzzy_post_handler",

+ .func = lua_fuzzy_add_post_handler,

+ rspamd_lua_add_metamethod(ctx->cfg->lua_state, "rspamd{worker}", &fuzzy_lua_reg);

+ rspamd_lua_add_metamethod(ctx->cfg->lua_state, "rspamd{worker}", &fuzzy_lua_reg);

+ rspamd_lua_run_postloads(ctx->cfg->lua_state, ctx->cfg, ctx->event_loop,

+ worker);

+ ev_loop(ctx->event_loop, 0);

+ rspamd_worker_block_signals();

+ ev_io_stop(ctx->event_loop, &ctx->peer_ev);

+ close(ctx->peer_fd);

+ msg_info_config("start another event loop to sync fuzzy storage");

+ if (rspamd_fuzzy_process_updates_queue(ctx, local_db_name, TRUE)) {

+ ev_loop(ctx->event_loop, 0);

+ msg_info_config("sync cycle is done");

+ msg_info_config("no need to sync");

+ rspamd_fuzzy_backend_close(ctx->backend);

+ g_array_free(ctx->updates_pending, TRUE);

+ rspamd_keypair_cache_destroy(ctx->keypair_cache);

+ rspamd_fuzzy_maybe_save_ratelimits(ctx);

+ rspamd_lru_hash_destroy(ctx->ratelimit_buckets);

+ luaL_unref(ctx->cfg->lua_state, LUA_REGISTRYINDEX, ctx->lua_pre_handler_cbref);

+ luaL_unref(ctx->cfg->lua_state, LUA_REGISTRYINDEX, ctx->lua_post_handler_cbref);

+ luaL_unref(ctx->cfg->lua_state, LUA_REGISTRYINDEX, ctx->lua_blacklist_cbref);

+ REF_RELEASE(ctx->cfg);

+ rspamd_log_close(worker->srv->logger);

+ rspamd_unset_crash_handler(worker->srv);

+ exit(EXIT_SUCCESS);