123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011101210131014101510161017101810191020102110221023102410251026102710281029103010311032103310341035103610371038103910401041104210431044104510461047104810491050105110521053105410551056105710581059106010611062106310641065106610671068106910701071107210731074107510761077107810791080108110821083108410851086108710881089109010911092109310941095109610971098109911001101110211031104110511061107110811091110111111121113111411151116111711181119112011211122112311241125112611271128112911301131113211331134113511361137113811391140114111421143114411451146114711481149115011511152115311541155115611571158115911601161116211631164116511661167116811691170117111721173117411751176117711781179118011811182118311841185118611871188118911901191119211931194119511961197119811991200120112021203120412051206120712081209121012111212121312141215121612171218121912201221122212231224122512261227122812291230123112321233123412351236123712381239124012411242124312441245124612471248124912501251125212531254125512561257125812591260126112621263126412651266126712681269127012711272127312741275127612771278127912801281128212831284128512861287128812891290129112921293129412951296129712981299130013011302130313041305130613071308130913101311131213131314131513161317131813191320132113221323132413251326132713281329133013311332133313341335133613371338133913401341134213431344134513461347134813491350135113521353135413551356135713581359136013611362136313641365136613671368136913701371137213731374137513761377137813791380138113821383138413851386138713881389139013911392139313941395139613971398139914001401140214031404140514061407140814091410141114121413141414151416141714181419142014211422142314241425142614271428142914301431143214331434143514361437143814391440144114421443144414451446144714481449145014511452145314541455145614571458145914601461146214631464146514661467146814691470147114721473147414751476147714781479148014811482148314841485148614871488148914901491149214931494149514961497149814991500150115021503150415051506150715081509151015111512151315141515151615171518151915201521152215231524152515261527152815291530153115321533153415351536153715381539154015411542154315441545154615471548154915501551155215531554155515561557155815591560156115621563156415651566156715681569157015711572157315741575157615771578157915801581158215831584158515861587158815891590159115921593159415951596159715981599160016011602160316041605160616071608160916101611161216131614161516161617161816191620162116221623162416251626162716281629163016311632163316341635163616371638163916401641164216431644164516461647164816491650165116521653165416551656165716581659166016611662166316641665166616671668166916701671167216731674167516761677167816791680168116821683168416851686168716881689169016911692169316941695169616971698169917001701170217031704170517061707170817091710171117121713171417151716171717181719172017211722172317241725172617271728172917301731173217331734173517361737173817391740174117421743174417451746174717481749175017511752175317541755175617571758175917601761176217631764176517661767176817691770177117721773177417751776177717781779178017811782178317841785178617871788178917901791179217931794179517961797179817991800180118021803180418051806180718081809181018111812181318141815181618171818181918201821182218231824182518261827182818291830183118321833183418351836183718381839184018411842184318441845184618471848184918501851185218531854185518561857185818591860186118621863186418651866186718681869187018711872187318741875187618771878187918801881188218831884188518861887188818891890189118921893189418951896189718981899190019011902190319041905190619071908190919101911191219131914191519161917191819191920192119221923192419251926192719281929193019311932193319341935193619371938193919401941194219431944194519461947194819491950195119521953195419551956195719581959196019611962196319641965196619671968196919701971197219731974197519761977197819791980198119821983198419851986198719881989199019911992199319941995199619971998199920002001200220032004200520062007200820092010201120122013201420152016201720182019202020212022202320242025202620272028202920302031203220332034203520362037203820392040204120422043204420452046204720482049205020512052205320542055205620572058205920602061206220632064206520662067206820692070207120722073207420752076207720782079208020812082208320842085208620872088208920902091209220932094209520962097209820992100210121022103210421052106210721082109211021112112211321142115211621172118211921202121212221232124212521262127212821292130213121322133213421352136213721382139214021412142214321442145214621472148214921502151215221532154215521562157215821592160216121622163216421652166216721682169217021712172217321742175217621772178217921802181218221832184218521862187218821892190219121922193219421952196219721982199220022012202220322042205220622072208220922102211221222132214221522162217221822192220222122222223222422252226222722282229223022312232223322342235223622372238223922402241224222432244224522462247224822492250225122522253225422552256225722582259226022612262226322642265226622672268226922702271227222732274227522762277227822792280228122822283228422852286228722882289229022912292229322942295229622972298229923002301230223032304230523062307230823092310231123122313231423152316231723182319232023212322232323242325232623272328232923302331233223332334233523362337233823392340234123422343234423452346234723482349235023512352235323542355235623572358235923602361236223632364236523662367236823692370237123722373237423752376237723782379238023812382238323842385238623872388238923902391239223932394239523962397239823992400240124022403240424052406240724082409241024112412241324142415241624172418241924202421242224232424242524262427 |
- /*-
- * Copyright 2016 Vsevolod Stakhov
- *
- * 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.
- */
- #include "config.h"
- #include "libutil/util.h"
- #include "libserver/maps/map.h"
- #include "libutil/upstream.h"
- #include "libserver/http/http_connection.h"
- #include "libserver/http/http_private.h"
- #include "libserver/protocol.h"
- #include "libserver/protocol_internal.h"
- #include "libserver/cfg_file.h"
- #include "libserver/url.h"
- #include "libserver/dns.h"
- #include "libmime/message.h"
- #include "rspamd.h"
- #include "libserver/worker_util.h"
- #include "worker_private.h"
- #include "lua/lua_common.h"
- #include "keypairs_cache.h"
- #include "libstat/stat_api.h"
- #include "ottery.h"
- #include "unix-std.h"
- #include "libserver/milter.h"
- #include "libserver/milter_internal.h"
- #include "libmime/lang_detection.h"
-
- #include <math.h>
-
- #ifdef HAVE_NETINET_TCP_H
- #include <netinet/tcp.h> /* for TCP_NODELAY */
- #endif
-
- #ifdef SYS_ZSTD
- #include "zstd.h"
- #else
- #include "contrib/zstd/zstd.h"
- #endif
-
- /* Rotate keys each minute by default */
- #define DEFAULT_ROTATION_TIME 60.0
- #define DEFAULT_RETRIES 5
-
- #define msg_err_session(...) rspamd_default_log_function(G_LOG_LEVEL_CRITICAL, \
- session->pool->tag.tagname, session->pool->tag.uid, \
- RSPAMD_LOG_FUNC, \
- __VA_ARGS__)
- #define msg_warn_session(...) rspamd_default_log_function(G_LOG_LEVEL_WARNING, \
- session->pool->tag.tagname, session->pool->tag.uid, \
- RSPAMD_LOG_FUNC, \
- __VA_ARGS__)
- #define msg_info_session(...) rspamd_default_log_function(G_LOG_LEVEL_INFO, \
- session->pool->tag.tagname, session->pool->tag.uid, \
- RSPAMD_LOG_FUNC, \
- __VA_ARGS__)
-
- #define msg_debug_session(...) rspamd_conditional_debug_fast(NULL, session->client_addr, \
- rspamd_proxy_log_id, "proxy", session->pool->tag.uid, \
- RSPAMD_LOG_FUNC, \
- __VA_ARGS__)
-
- INIT_LOG_MODULE(proxy)
-
- gpointer init_rspamd_proxy(struct rspamd_config *cfg);
- void start_rspamd_proxy(struct rspamd_worker *worker);
-
- worker_t rspamd_proxy_worker = {
- "rspamd_proxy", /* Name */
- init_rspamd_proxy, /* Init function */
- start_rspamd_proxy, /* Start function */
- RSPAMD_WORKER_HAS_SOCKET | RSPAMD_WORKER_KILLABLE | RSPAMD_WORKER_SCANNER,
- RSPAMD_WORKER_SOCKET_TCP, /* TCP socket */
- RSPAMD_WORKER_VER};
-
- struct rspamd_http_upstream {
- gchar *name;
- gchar *settings_id;
- struct upstream_list *u;
- struct rspamd_cryptobox_pubkey *key;
- gdouble timeout;
- gint parser_from_ref;
- gint parser_to_ref;
- gboolean local;
- gboolean self_scan;
- gboolean compress;
- };
-
- struct rspamd_http_mirror {
- gchar *name;
- gchar *settings_id;
- struct upstream_list *u;
- struct rspamd_cryptobox_pubkey *key;
- gdouble prob;
- gdouble timeout;
- gint parser_from_ref;
- gint parser_to_ref;
- gboolean local;
- gboolean compress;
- };
-
- static const guint64 rspamd_rspamd_proxy_magic = 0xcdeb4fd1fc351980ULL;
-
- struct rspamd_proxy_ctx {
- guint64 magic;
- /* Events base */
- struct ev_loop *event_loop;
- /* DNS resolver */
- struct rspamd_dns_resolver *resolver;
- /* Config */
- struct rspamd_config *cfg;
- /* END OF COMMON PART */
- gdouble timeout;
- /* Encryption key for clients */
- struct rspamd_cryptobox_keypair *key;
- /* HTTP context */
- struct rspamd_http_context *http_ctx;
- /* Upstreams to use */
- GHashTable *upstreams;
- /* Mirrors to send traffic to */
- GPtrArray *mirrors;
- /* Default upstream */
- struct rspamd_http_upstream *default_upstream;
- lua_State *lua_state;
- /* Array of callback functions called on end of scan to compare results */
- GArray *cmp_refs;
- /* Maximum count for retries */
- guint max_retries;
- /* If we have self_scanning backends, we need to work as a normal worker */
- gboolean has_self_scan;
- /* It is not HTTP but milter proxy */
- gboolean milter;
- /* Discard messages instead of rejecting them */
- gboolean discard_on_reject;
- /* Quarantine messages instead of rejecting them */
- gboolean quarantine_on_reject;
- /* Milter spam header */
- gchar *spam_header;
- /* CA name that can be used for client certificates */
- gchar *client_ca_name;
- /* Milter rejection message */
- gchar *reject_message;
- /* Sessions cache */
- void *sessions_cache;
- struct rspamd_milter_context milter_ctx;
- /* Language detector */
- struct rspamd_lang_detector *lang_det;
- gdouble task_timeout;
- };
-
- enum rspamd_backend_flags {
- RSPAMD_BACKEND_REPLIED = 1 << 0,
- RSPAMD_BACKEND_CLOSED = 1 << 1,
- RSPAMD_BACKEND_PARSED = 1 << 2,
- };
-
- struct rspamd_proxy_session;
-
- struct rspamd_proxy_backend_connection {
- const gchar *name;
- struct rspamd_cryptobox_keypair *local_key;
- struct rspamd_cryptobox_pubkey *remote_key;
- struct upstream *up;
- struct rspamd_http_connection *backend_conn;
- ucl_object_t *results;
- const gchar *err;
- struct rspamd_proxy_session *s;
- gint backend_sock;
- ev_tstamp timeout;
- enum rspamd_backend_flags flags;
- gint parser_from_ref;
- gint parser_to_ref;
- struct rspamd_task *task;
- };
-
- enum rspamd_proxy_legacy_support {
- LEGACY_SUPPORT_NO = 0,
- LEGACY_SUPPORT_RSPAMC,
- LEGACY_SUPPORT_SPAMC
- };
-
- struct rspamd_proxy_session {
- struct rspamd_worker *worker;
- rspamd_mempool_t *pool;
- struct rspamd_proxy_ctx *ctx;
- rspamd_inet_addr_t *client_addr;
- struct rspamd_http_connection *client_conn;
- struct rspamd_milter_session *client_milter_conn;
- struct rspamd_http_upstream *backend;
- gpointer map;
- gchar *fname;
- gpointer shmem_ref;
- struct rspamd_proxy_backend_connection *master_conn;
- struct rspamd_http_message *client_message;
- GPtrArray *mirror_conns;
- gsize map_len;
- gint client_sock;
- enum rspamd_proxy_legacy_support legacy_support;
- gint retries;
- ref_entry_t ref;
- };
-
- static gboolean proxy_send_master_message(struct rspamd_proxy_session *session);
-
- static GQuark
- rspamd_proxy_quark(void)
- {
- return g_quark_from_static_string("rspamd-proxy");
- }
-
- static gboolean
- rspamd_proxy_parse_lua_parser(lua_State *L, const ucl_object_t *obj,
- gint *ref_from, gint *ref_to, GError **err)
- {
- const gchar *lua_script;
- gsize slen;
- gint err_idx, ref_idx;
- gboolean has_ref = FALSE;
-
- g_assert(obj != NULL);
- g_assert(ref_from != NULL);
- g_assert(ref_to != NULL);
-
- *ref_from = -1;
- *ref_to = -1;
-
- lua_script = ucl_object_tolstring(obj, &slen);
- lua_pushcfunction(L, &rspamd_lua_traceback);
- err_idx = lua_gettop(L);
-
- /* Load data */
- if (luaL_loadbuffer(L, lua_script, slen, "proxy parser") != 0) {
- g_set_error(err,
- rspamd_proxy_quark(),
- EINVAL,
- "cannot load lua parser script: %s",
- lua_tostring(L, -1));
- lua_settop(L, 0); /* Error function */
-
- return FALSE;
- }
-
- /* Now do it */
- if (lua_pcall(L, 0, 1, err_idx) != 0) {
- g_set_error(err,
- rspamd_proxy_quark(),
- EINVAL,
- "cannot init lua parser script: %s",
- lua_tostring(L, -1));
- lua_settop(L, 0);
-
- return FALSE;
- }
-
- if (lua_istable(L, -1)) {
- /*
- * We have a table, so we check for two keys:
- * 'from' -> function
- * 'to' -> function
- *
- * From converts parent request to a client one
- * To converts client request to a parent one
- */
- lua_pushstring(L, "from");
- lua_gettable(L, -2);
-
- if (lua_isfunction(L, -1)) {
- ref_idx = luaL_ref(L, LUA_REGISTRYINDEX);
- *ref_from = ref_idx;
- has_ref = TRUE;
- }
-
- lua_pushstring(L, "to");
- lua_gettable(L, -2);
-
- if (lua_isfunction(L, -1)) {
- ref_idx = luaL_ref(L, LUA_REGISTRYINDEX);
- *ref_to = ref_idx;
- has_ref = TRUE;
- }
- }
- else if (!lua_isfunction(L, -1)) {
- g_set_error(err,
- rspamd_proxy_quark(),
- EINVAL,
- "cannot init lua parser script: "
- "must return function");
- lua_settop(L, 0);
-
- return FALSE;
- }
- else {
- /* Just parser from protocol */
- ref_idx = luaL_ref(L, LUA_REGISTRYINDEX);
- *ref_from = ref_idx;
- lua_settop(L, 0);
- has_ref = TRUE;
- }
-
- return has_ref;
- }
-
- static gboolean
- rspamd_proxy_parse_upstream(rspamd_mempool_t *pool,
- const ucl_object_t *obj,
- gpointer ud,
- struct rspamd_rcl_section *section,
- GError **err)
- {
- const ucl_object_t *elt;
- struct rspamd_http_upstream *up = NULL;
- struct rspamd_proxy_ctx *ctx;
- struct rspamd_rcl_struct_parser *pd = ud;
- lua_State *L;
-
- ctx = pd->user_struct;
- L = ctx->lua_state;
-
- if (ucl_object_type(obj) != UCL_OBJECT) {
- if (ucl_object_type(obj) != UCL_NULL) {
- msg_info_pool_check("upstream %s is disabled by setting it to NULL",
- ucl_object_key(obj));
- return TRUE;
- }
-
- g_set_error(err, rspamd_proxy_quark(), 100,
- "upstream option must be an object");
-
- return FALSE;
- }
-
- if (!rspamd_config_is_enabled_from_ucl(pool, obj)) {
- /* Upstream is valid but disabled */
- msg_info_pool_check("upstream %s is disabled",
- ucl_object_lookup(obj, "name") ? ucl_object_tostring(ucl_object_lookup(obj, "name")) : ucl_object_key(obj));
- return TRUE;
- }
-
- up = rspamd_mempool_alloc0(pool, sizeof(*up));
-
- elt = ucl_object_lookup(obj, "name");
- if (elt == NULL) {
-
- if (ucl_object_key(obj)) {
- if (strcmp(ucl_object_key(obj), "upstream") == 0) {
- /* Iterate over the object and find upstream elements */
- ucl_object_iter_t it = NULL;
- const ucl_object_t *cur;
- gboolean ret = TRUE;
-
- while ((cur = ucl_object_iterate(obj, &it, true)) != NULL) {
- if (!rspamd_proxy_parse_upstream(pool, cur, ud,
- section, err)) {
- ret = FALSE;
- }
- }
-
- return ret;
- }
- else {
- /* Inside upstream */
- up->name = rspamd_mempool_strdup(pool, ucl_object_key(obj));
- }
- }
- else {
- g_set_error(err, rspamd_proxy_quark(), 100,
- "upstream option must have some name definition");
-
- return FALSE;
- }
- }
- else {
- up->name = rspamd_mempool_strdup(pool, ucl_object_tostring(elt));
- }
-
- up->parser_from_ref = -1;
- up->parser_to_ref = -1;
- up->timeout = ctx->timeout;
-
- elt = ucl_object_lookup(obj, "key");
- if (elt != NULL) {
- up->key = rspamd_pubkey_from_base32(ucl_object_tostring(elt), 0,
- RSPAMD_KEYPAIR_KEX, RSPAMD_CRYPTOBOX_MODE_25519);
-
- if (up->key == NULL) {
- g_set_error(err, rspamd_proxy_quark(), 100,
- "cannot read upstream key");
-
- goto err;
- }
-
- rspamd_mempool_add_destructor(pool,
- (rspamd_mempool_destruct_t) rspamd_pubkey_unref, up->key);
- }
-
- elt = ucl_object_lookup(obj, "self_scan");
- if (elt && ucl_object_toboolean(elt)) {
- up->self_scan = TRUE;
- ctx->has_self_scan = TRUE;
- }
-
- elt = ucl_object_lookup_any(obj, "compress", "compression", NULL);
- if (elt && ucl_object_toboolean(elt)) {
- up->compress = TRUE;
- }
-
- elt = ucl_object_lookup(obj, "hosts");
-
- if (elt == NULL && !up->self_scan) {
- g_set_error(err, rspamd_proxy_quark(), 100,
- "upstream option must have some hosts definition");
-
- goto err;
- }
-
- if (elt) {
- up->u = rspamd_upstreams_create(ctx->cfg->ups_ctx);
-
- if (!rspamd_upstreams_from_ucl(up->u, elt, 11333, NULL)) {
- g_set_error(err, rspamd_proxy_quark(), 100,
- "upstream has bad hosts definition");
-
- goto err;
- }
-
- rspamd_mempool_add_destructor(pool,
- (rspamd_mempool_destruct_t) rspamd_upstreams_destroy, up->u);
- }
-
- elt = ucl_object_lookup(obj, "default");
- if (elt) {
- if (ucl_object_toboolean(elt)) {
- ctx->default_upstream = up;
- }
- }
- else if (up->self_scan) {
- ctx->default_upstream = up;
- }
-
-
- elt = ucl_object_lookup(obj, "local");
- if (elt && ucl_object_toboolean(elt)) {
- up->local = TRUE;
- }
-
- elt = ucl_object_lookup(obj, "timeout");
- if (elt) {
- ucl_object_todouble_safe(elt, &up->timeout);
- }
-
- elt = ucl_object_lookup_any(obj, "settings", "settings_id", NULL);
- if (elt && ucl_object_type(elt) == UCL_STRING) {
- up->settings_id = rspamd_mempool_strdup(pool, ucl_object_tostring(elt));
- }
-
- /*
- * Accept lua function here in form
- * fun :: String -> UCL
- */
- elt = ucl_object_lookup(obj, "parser");
- if (elt) {
- if (!rspamd_proxy_parse_lua_parser(L, elt, &up->parser_from_ref,
- &up->parser_to_ref, err)) {
- goto err;
- }
-
- rspamd_lua_add_ref_dtor(L, pool, up->parser_from_ref);
- rspamd_lua_add_ref_dtor(L, pool, up->parser_to_ref);
- }
-
- g_hash_table_insert(ctx->upstreams, up->name, up);
-
- return TRUE;
-
- err:
- return FALSE;
- }
-
- static gboolean
- rspamd_proxy_parse_mirror(rspamd_mempool_t *pool,
- const ucl_object_t *obj,
- gpointer ud,
- struct rspamd_rcl_section *section,
- GError **err)
- {
- const ucl_object_t *elt;
- struct rspamd_http_mirror *up = NULL;
- struct rspamd_proxy_ctx *ctx;
- struct rspamd_rcl_struct_parser *pd = ud;
- lua_State *L;
-
- ctx = pd->user_struct;
- L = ctx->lua_state;
-
- if (ucl_object_type(obj) != UCL_OBJECT) {
- if (ucl_object_type(obj) != UCL_NULL) {
- msg_info_pool_check("mirror %s is disabled by setting it to NULL",
- ucl_object_key(obj));
- return TRUE;
- }
-
- g_set_error(err, rspamd_proxy_quark(), 100,
- "mirror option must be an object");
-
- return FALSE;
- }
-
- if (!rspamd_config_is_enabled_from_ucl(pool, obj)) {
- /* Upstream is valid but disabled */
- msg_info_pool_check("mirror %s is disabled",
- ucl_object_lookup(obj, "name") ? ucl_object_tostring(ucl_object_lookup(obj, "name")) : ucl_object_key(obj));
- return TRUE;
- }
-
- up = rspamd_mempool_alloc0(pool, sizeof(*up));
-
- elt = ucl_object_lookup(obj, "name");
- if (elt == NULL) {
-
- if (ucl_object_key(obj)) {
- if (strcmp(ucl_object_key(obj), "mirror") == 0) {
- /* Iterate over the object and find upstream elements */
- ucl_object_iter_t it = NULL;
- const ucl_object_t *cur;
- gboolean ret = TRUE;
-
- while ((cur = ucl_object_iterate(obj, &it, true)) != NULL) {
- if (!rspamd_proxy_parse_mirror(pool, cur, ud,
- section, err)) {
- ret = FALSE;
- }
- }
-
- return ret;
- }
- else {
- /* Inside upstream */
- up->name = rspamd_mempool_strdup(pool, ucl_object_key(obj));
- }
- }
- else {
- g_set_error(err, rspamd_proxy_quark(), 100,
- "mirror option must have some name definition");
-
- return FALSE;
- }
- }
- else {
- up->name = rspamd_mempool_strdup(pool, ucl_object_tostring(elt));
- }
-
- up->parser_to_ref = -1;
- up->parser_from_ref = -1;
- up->timeout = ctx->timeout;
-
- elt = ucl_object_lookup(obj, "key");
- if (elt != NULL) {
- up->key = rspamd_pubkey_from_base32(ucl_object_tostring(elt), 0,
- RSPAMD_KEYPAIR_KEX, RSPAMD_CRYPTOBOX_MODE_25519);
-
- if (up->key == NULL) {
- g_set_error(err, rspamd_proxy_quark(), 100,
- "cannot read mirror key");
-
- goto err;
- }
-
- rspamd_mempool_add_destructor(pool,
- (rspamd_mempool_destruct_t) rspamd_pubkey_unref, up->key);
- }
-
- elt = ucl_object_lookup(obj, "hosts");
-
- if (elt == NULL) {
- g_set_error(err, rspamd_proxy_quark(), 100,
- "mirror option must have some hosts definition");
-
- goto err;
- }
-
- up->u = rspamd_upstreams_create(ctx->cfg->ups_ctx);
- if (!rspamd_upstreams_from_ucl(up->u, elt, 11333, NULL)) {
- g_set_error(err, rspamd_proxy_quark(), 100,
- "mirror has bad hosts definition");
-
- goto err;
- }
-
- rspamd_mempool_add_destructor(pool,
- (rspamd_mempool_destruct_t) rspamd_upstreams_destroy, up->u);
-
- elt = ucl_object_lookup_any(obj, "probability", "prob", NULL);
- if (elt) {
- up->prob = ucl_object_todouble(elt);
- }
- else {
- up->prob = 1.0;
- }
-
- elt = ucl_object_lookup(obj, "local");
- if (elt && ucl_object_toboolean(elt)) {
- up->local = TRUE;
- }
-
- elt = ucl_object_lookup_any(obj, "compress", "compression", NULL);
- if (elt && ucl_object_toboolean(elt)) {
- up->compress = TRUE;
- }
-
- elt = ucl_object_lookup(obj, "timeout");
- if (elt) {
- ucl_object_todouble_safe(elt, &up->timeout);
- }
-
- /*
- * Accept lua function here in form
- * fun :: String -> UCL
- */
- elt = ucl_object_lookup(obj, "parser");
- if (elt) {
- if (!rspamd_proxy_parse_lua_parser(L, elt, &up->parser_from_ref,
- &up->parser_to_ref, err)) {
- goto err;
- }
-
- rspamd_lua_add_ref_dtor(L, pool, up->parser_from_ref);
- rspamd_lua_add_ref_dtor(L, pool, up->parser_to_ref);
- }
-
- elt = ucl_object_lookup_any(obj, "settings", "settings_id", NULL);
- if (elt && ucl_object_type(elt) == UCL_STRING) {
- up->settings_id = rspamd_mempool_strdup(pool, ucl_object_tostring(elt));
- }
-
- g_ptr_array_add(ctx->mirrors, up);
-
- return TRUE;
-
- err:
-
- return FALSE;
- }
-
- static gboolean
- rspamd_proxy_parse_script(rspamd_mempool_t *pool,
- const ucl_object_t *obj,
- gpointer ud,
- struct rspamd_rcl_section *section,
- GError **err)
- {
- struct rspamd_proxy_ctx *ctx;
- struct rspamd_rcl_struct_parser *pd = ud;
- lua_State *L;
- const gchar *lua_script;
- gsize slen;
- gint err_idx, ref_idx;
- struct stat st;
-
- ctx = pd->user_struct;
- L = ctx->lua_state;
-
- if (ucl_object_type(obj) != UCL_STRING) {
- g_set_error(err, rspamd_proxy_quark(), 100,
- "script option must be a string with file or lua chunk");
-
- return FALSE;
- }
-
- lua_script = ucl_object_tolstring(obj, &slen);
- lua_pushcfunction(L, &rspamd_lua_traceback);
- err_idx = lua_gettop(L);
-
- if (stat(lua_script, &st) != -1) {
- /* Load file */
- if (luaL_loadfile(L, lua_script) != 0) {
- g_set_error(err,
- rspamd_proxy_quark(),
- EINVAL,
- "cannot load lua parser script: %s",
- lua_tostring(L, -1));
- lua_settop(L, 0); /* Error function */
-
- goto err;
- }
- }
- else {
- /* Load data directly */
- if (luaL_loadbuffer(L, lua_script, slen, "proxy parser") != 0) {
- g_set_error(err,
- rspamd_proxy_quark(),
- EINVAL,
- "cannot load lua parser script: %s",
- lua_tostring(L, -1));
- lua_settop(L, 0); /* Error function */
-
- goto err;
- }
- }
-
- /* Now do it */
- if (lua_pcall(L, 0, 1, err_idx) != 0) {
- g_set_error(err,
- rspamd_proxy_quark(),
- EINVAL,
- "cannot init lua parser script: %s",
- lua_tostring(L, -1));
- lua_settop(L, 0);
-
- goto err;
- }
-
- if (!lua_isfunction(L, -1)) {
- g_set_error(err,
- rspamd_proxy_quark(),
- EINVAL,
- "cannot init lua parser script: "
- "must return function, %s returned",
- lua_typename(L, lua_type(L, -1)));
- lua_settop(L, 0);
-
- goto err;
- }
-
- ref_idx = luaL_ref(L, LUA_REGISTRYINDEX);
- lua_settop(L, 0);
- g_array_append_val(ctx->cmp_refs, ref_idx);
-
- return TRUE;
-
- err:
- return FALSE;
- }
-
- gpointer
- init_rspamd_proxy(struct rspamd_config *cfg)
- {
- struct rspamd_proxy_ctx *ctx;
- GQuark type;
-
- type = g_quark_try_string("rspamd_proxy");
-
- ctx = rspamd_mempool_alloc0(cfg->cfg_pool,
- sizeof(struct rspamd_proxy_ctx));
- ctx->magic = rspamd_rspamd_proxy_magic;
- ctx->timeout = 120.0;
- ctx->upstreams = g_hash_table_new(rspamd_strcase_hash, rspamd_strcase_equal);
- rspamd_mempool_add_destructor(cfg->cfg_pool,
- (rspamd_mempool_destruct_t) g_hash_table_unref, ctx->upstreams);
- ctx->mirrors = g_ptr_array_new();
- rspamd_mempool_add_destructor(cfg->cfg_pool,
- (rspamd_mempool_destruct_t) rspamd_ptr_array_free_hard, ctx->mirrors);
- ctx->cfg = cfg;
- ctx->lua_state = cfg->lua_state;
- ctx->cmp_refs = g_array_new(FALSE, FALSE, sizeof(gint));
- rspamd_mempool_add_destructor(cfg->cfg_pool,
- (rspamd_mempool_destruct_t) rspamd_array_free_hard, ctx->cmp_refs);
- ctx->max_retries = DEFAULT_RETRIES;
- ctx->spam_header = RSPAMD_MILTER_SPAM_HEADER;
-
- rspamd_rcl_register_worker_option(cfg,
- type,
- "timeout",
- rspamd_rcl_parse_struct_time,
- ctx,
- G_STRUCT_OFFSET(struct rspamd_proxy_ctx, timeout),
- RSPAMD_CL_FLAG_TIME_FLOAT,
- "IO timeout");
- rspamd_rcl_register_worker_option(cfg,
- type,
- "keypair",
- rspamd_rcl_parse_struct_keypair,
- ctx,
- G_STRUCT_OFFSET(struct rspamd_proxy_ctx, key),
- 0,
- "Server's keypair");
- rspamd_rcl_register_worker_option(cfg,
- type,
- "upstream",
- rspamd_proxy_parse_upstream,
- ctx,
- 0,
- 0,
- "List of upstreams");
- rspamd_rcl_register_worker_option(cfg,
- type,
- "mirror",
- rspamd_proxy_parse_mirror,
- ctx,
- 0,
- RSPAMD_CL_FLAG_MULTIPLE,
- "List of mirrors");
-
- rspamd_rcl_register_worker_option(cfg,
- type,
- "script",
- rspamd_proxy_parse_script,
- ctx,
- 0,
- RSPAMD_CL_FLAG_MULTIPLE,
- "Compare script to be executed");
- rspamd_rcl_register_worker_option(cfg,
- type,
- "max_retries",
- rspamd_rcl_parse_struct_integer,
- ctx,
- G_STRUCT_OFFSET(struct rspamd_proxy_ctx, max_retries),
- RSPAMD_CL_FLAG_UINT,
- "Maximum number of retries for master connection");
- rspamd_rcl_register_worker_option(cfg,
- type,
- "milter",
- rspamd_rcl_parse_struct_boolean,
- ctx,
- G_STRUCT_OFFSET(struct rspamd_proxy_ctx, milter),
- 0,
- "Accept milter connections, not HTTP");
- rspamd_rcl_register_worker_option(cfg,
- type,
- "discard_on_reject",
- rspamd_rcl_parse_struct_boolean,
- ctx,
- G_STRUCT_OFFSET(struct rspamd_proxy_ctx, discard_on_reject),
- 0,
- "Tell MTA to discard rejected messages silently");
- rspamd_rcl_register_worker_option(cfg,
- type,
- "quarantine_on_reject",
- rspamd_rcl_parse_struct_boolean,
- ctx,
- G_STRUCT_OFFSET(struct rspamd_proxy_ctx, quarantine_on_reject),
- 0,
- "Tell MTA to quarantine rejected messages");
- rspamd_rcl_register_worker_option(cfg,
- type,
- "spam_header",
- rspamd_rcl_parse_struct_string,
- ctx,
- G_STRUCT_OFFSET(struct rspamd_proxy_ctx, spam_header),
- 0,
- "Use the specific spam header (default: X-Spam)");
- rspamd_rcl_register_worker_option(cfg,
- type,
- "client_ca_name",
- rspamd_rcl_parse_struct_string,
- ctx,
- G_STRUCT_OFFSET(struct rspamd_proxy_ctx, client_ca_name),
- 0,
- "Allow certificates issued by this CA to be treated as client certificates");
- rspamd_rcl_register_worker_option(cfg,
- type,
- "reject_message",
- rspamd_rcl_parse_struct_string,
- ctx,
- G_STRUCT_OFFSET(struct rspamd_proxy_ctx, reject_message),
- 0,
- "Use custom rejection message");
-
- return ctx;
- }
-
- static void
- proxy_backend_close_connection(struct rspamd_proxy_backend_connection *conn)
- {
- if (conn && !(conn->flags & RSPAMD_BACKEND_CLOSED)) {
- if (conn->backend_conn) {
- rspamd_http_connection_reset(conn->backend_conn);
- rspamd_http_connection_unref(conn->backend_conn);
- close(conn->backend_sock);
- }
-
- conn->flags |= RSPAMD_BACKEND_CLOSED;
- }
- }
-
- static gboolean
- proxy_backend_parse_results(struct rspamd_proxy_session *session,
- struct rspamd_proxy_backend_connection *conn,
- lua_State *L, gint parser_ref,
- struct rspamd_http_message *msg,
- goffset *body_offset,
- const rspamd_ftok_t *ct)
- {
- struct ucl_parser *parser;
- gint err_idx;
- const gchar *in = msg->body_buf.begin;
- gsize inlen = msg->body_buf.len;
- const rspamd_ftok_t *offset_hdr;
-
- if (inlen == 0 || in == NULL) {
- return FALSE;
- }
-
- offset_hdr = rspamd_http_message_find_header(msg, MESSAGE_OFFSET_HEADER);
-
- if (offset_hdr) {
- gulong val;
-
- if (rspamd_strtoul(offset_hdr->begin, offset_hdr->len, &val) && val < inlen) {
-
- if (body_offset) {
- *body_offset = val;
- }
- inlen = val;
- }
- }
-
- if (parser_ref != -1) {
- /* Call parser function */
- lua_pushcfunction(L, &rspamd_lua_traceback);
- err_idx = lua_gettop(L);
-
- lua_rawgeti(L, LUA_REGISTRYINDEX, parser_ref);
- /* XXX: copies all data */
- lua_pushlstring(L, in, inlen);
-
- if (lua_pcall(L, 1, 1, err_idx) != 0) {
- msg_err_session(
- "cannot run lua parser script: %s",
- lua_tostring(L, -1));
- lua_settop(L, 0);
-
- return FALSE;
- }
-
- conn->results = ucl_object_lua_import(L, -1);
- lua_settop(L, 0);
- }
- else {
- rspamd_ftok_t json_ct;
- RSPAMD_FTOK_ASSIGN(&json_ct, "application/json");
-
- if (ct && rspamd_ftok_casecmp(ct, &json_ct) == 0) {
- parser = ucl_parser_new(0);
-
- if (!ucl_parser_add_chunk(parser, in, inlen)) {
- gchar *encoded;
-
- encoded = rspamd_encode_base64(in, inlen, 0, NULL);
- msg_err_session("cannot parse input: %s", ucl_parser_get_error(
- parser));
- msg_err_session("input encoded: %s", encoded);
- ucl_parser_free(parser);
- g_free(encoded);
-
- return FALSE;
- }
-
- conn->results = ucl_parser_get_object(parser);
- ucl_parser_free(parser);
- }
- }
-
- return TRUE;
- }
-
- static void
- proxy_call_cmp_script(struct rspamd_proxy_session *session, gint cbref)
- {
- gint err_idx;
- guint i;
- struct rspamd_proxy_backend_connection *conn;
- lua_State *L;
-
- L = session->ctx->lua_state;
- lua_pushcfunction(L, &rspamd_lua_traceback);
- err_idx = lua_gettop(L);
-
- lua_rawgeti(L, LUA_REGISTRYINDEX, cbref);
-
- lua_createtable(L, 0, session->mirror_conns->len + 1);
- /* Now push master results */
- if (session->master_conn && session->master_conn->results) {
- lua_pushstring(L, "master");
- ucl_object_push_lua(L, session->master_conn->results, true);
- lua_settable(L, -3);
- }
- else {
- lua_pushstring(L, "master");
- lua_pushstring(L, "no results");
- lua_settable(L, -3);
- }
-
- for (i = 0; i < session->mirror_conns->len; i++) {
- conn = g_ptr_array_index(session->mirror_conns, i);
-
- if (conn->results) {
- lua_pushstring(L, conn->name);
- ucl_object_push_lua(L, conn->results, true);
- lua_settable(L, -3);
- }
- else {
- lua_pushstring(L, conn->name);
- lua_pushstring(L, conn->err ? conn->err : "unknown error");
- lua_settable(L, -3);
- }
- }
-
- gchar log_tag[RSPAMD_LOG_ID_LEN + 1];
- rspamd_strlcpy(log_tag, session->pool->tag.uid, sizeof(log_tag));
- lua_pushstring(L, log_tag);
-
- if (lua_pcall(L, 2, 0, err_idx) != 0) {
- msg_err_session(
- "cannot run lua compare script: %s",
- lua_tostring(L, -1));
- }
-
- lua_settop(L, err_idx - 1);
- }
-
- static void
- proxy_session_dtor(struct rspamd_proxy_session *session)
- {
- guint i;
- gint cbref;
- struct rspamd_proxy_backend_connection *conn;
-
- if (session->master_conn && session->master_conn->results) {
- for (i = 0; i < session->ctx->cmp_refs->len; i++) {
- cbref = g_array_index(session->ctx->cmp_refs, gint, i);
- proxy_call_cmp_script(session, cbref);
- }
- }
-
- if (session->master_conn) {
- proxy_backend_close_connection(session->master_conn);
- }
-
- if (session->client_milter_conn) {
- rspamd_milter_session_unref(session->client_milter_conn);
- }
- else if (session->client_conn) {
- rspamd_http_connection_reset(session->client_conn);
- rspamd_http_connection_unref(session->client_conn);
- }
-
- if (session->map && session->map_len) {
- munmap(session->map, session->map_len);
- }
-
- for (i = 0; i < session->mirror_conns->len; i++) {
- conn = g_ptr_array_index(session->mirror_conns, i);
-
- if (!(conn->flags & RSPAMD_BACKEND_CLOSED)) {
- proxy_backend_close_connection(conn);
- }
-
- if (conn->results) {
- ucl_object_unref(conn->results);
- }
- }
-
- if (session->master_conn) {
- if (session->master_conn->results) {
- ucl_object_unref(session->master_conn->results);
- }
-
- if (session->master_conn->task) {
- rspamd_session_destroy(session->master_conn->task->s);
- }
- }
-
- g_ptr_array_free(session->mirror_conns, TRUE);
- rspamd_http_message_shmem_unref(session->shmem_ref);
- rspamd_http_message_unref(session->client_message);
-
- if (session->client_addr) {
- rspamd_inet_address_free(session->client_addr);
- }
-
- if (session->client_sock != -1) {
- close(session->client_sock);
- }
-
- if (session->ctx->sessions_cache) {
- rspamd_worker_session_cache_remove(session->ctx->sessions_cache,
- session);
- }
-
- if (session->pool) {
- rspamd_mempool_delete(session->pool);
- }
-
- g_free(session);
- }
-
- static void
- proxy_request_compress(struct rspamd_http_message *msg)
- {
- guint flags;
- ZSTD_CCtx *zctx;
- rspamd_fstring_t *body;
- const gchar *in;
- gsize inlen;
-
- flags = rspamd_http_message_get_flags(msg);
-
- if (!rspamd_http_message_find_header(msg, COMPRESSION_HEADER)) {
- if ((flags & RSPAMD_HTTP_FLAG_SHMEM) ||
- !(flags & RSPAMD_HTTP_FLAG_HAS_BODY)) {
- /* Cannot compress shared or empty message */
- return;
- }
-
- in = rspamd_http_message_get_body(msg, &inlen);
-
- if (in == NULL || inlen == 0) {
- return;
- }
-
- body = rspamd_fstring_sized_new(ZSTD_compressBound(inlen));
- zctx = ZSTD_createCCtx();
- body->len = ZSTD_compressCCtx(zctx, body->str, body->allocated,
- in, inlen, 1);
-
- if (ZSTD_isError(body->len)) {
- msg_err("compression error");
- rspamd_fstring_free(body);
- ZSTD_freeCCtx(zctx);
-
- return;
- }
-
- ZSTD_freeCCtx(zctx);
- rspamd_http_message_set_body_from_fstring_steal(msg, body);
- rspamd_http_message_add_header(msg, COMPRESSION_HEADER, "zstd");
- }
- }
-
- static void
- proxy_request_decompress(struct rspamd_http_message *msg)
- {
- rspamd_fstring_t *body;
- const gchar *in;
- gsize inlen, outlen, r;
- ZSTD_DStream *zstream;
- ZSTD_inBuffer zin;
- ZSTD_outBuffer zout;
-
- if (rspamd_http_message_find_header(msg, COMPRESSION_HEADER)) {
- in = rspamd_http_message_get_body(msg, &inlen);
-
- if (in == NULL || inlen == 0) {
- return;
- }
-
- zstream = ZSTD_createDStream();
- ZSTD_initDStream(zstream);
-
- zin.pos = 0;
- zin.src = in;
- zin.size = inlen;
-
- if ((outlen = ZSTD_getDecompressedSize(zin.src, zin.size)) == 0) {
- outlen = ZSTD_DStreamOutSize();
- }
-
- body = rspamd_fstring_sized_new(outlen);
- zout.dst = body->str;
- zout.pos = 0;
- zout.size = outlen;
-
- while (zin.pos < zin.size) {
- r = ZSTD_decompressStream(zstream, &zout, &zin);
-
- if (ZSTD_isError(r)) {
- msg_err("Decompression error: %s", ZSTD_getErrorName(r));
- ZSTD_freeDStream(zstream);
- rspamd_fstring_free(body);
-
- return;
- }
-
- if (zout.pos == zout.size) {
- /* We need to extend output buffer */
- zout.size = zout.size * 2 + 1;
- body = rspamd_fstring_grow(body, zout.size);
- zout.size = body->allocated;
- zout.dst = body->str;
- }
- }
-
- body->len = zout.pos;
- ZSTD_freeDStream(zstream);
- rspamd_http_message_set_body_from_fstring_steal(msg, body);
- rspamd_http_message_remove_header(msg, COMPRESSION_HEADER);
- }
- }
-
- static struct rspamd_proxy_session *
- proxy_session_refresh(struct rspamd_proxy_session *session)
- {
- struct rspamd_proxy_session *nsession;
-
- nsession = g_malloc0(sizeof(*nsession));
- nsession->client_milter_conn = session->client_milter_conn;
- session->client_milter_conn = NULL;
- rspamd_milter_update_userdata(nsession->client_milter_conn,
- nsession);
- nsession->client_addr = session->client_addr;
- session->client_addr = NULL;
- nsession->ctx = session->ctx;
- nsession->worker = session->worker;
- nsession->pool = rspamd_mempool_new(rspamd_mempool_suggest_size(), "proxy", 0);
- nsession->client_sock = session->client_sock;
- session->client_sock = -1;
- nsession->mirror_conns = g_ptr_array_sized_new(nsession->ctx->mirrors->len);
-
- REF_INIT_RETAIN(nsession, proxy_session_dtor);
-
- if (nsession->ctx->sessions_cache) {
- rspamd_worker_session_cache_add(nsession->ctx->sessions_cache,
- nsession->pool->tag.uid, &nsession->ref.refcount, nsession);
- }
-
- return nsession;
- }
-
- static gboolean
- proxy_check_file(struct rspamd_http_message *msg,
- struct rspamd_proxy_session *session)
- {
- const rspamd_ftok_t *tok, *key_tok;
- rspamd_ftok_t srch;
- gchar *file_str;
- GHashTable *query_args;
- GHashTableIter it;
- gpointer k, v;
- struct http_parser_url u;
- rspamd_fstring_t *new_url;
-
- tok = rspamd_http_message_find_header(msg, "File");
-
- if (tok) {
- file_str = rspamd_mempool_ftokdup(session->pool, tok);
- session->map = rspamd_file_xmap(file_str, PROT_READ, &session->map_len,
- TRUE);
-
- if (session->map == NULL) {
- if (session->map_len != 0) {
- msg_err_session("cannot map %s: %s", file_str,
- strerror(errno));
-
- return FALSE;
- }
- }
- /* Remove header after processing */
- rspamd_http_message_remove_header(msg, "File");
- session->fname = file_str;
- }
- else {
- /* Need to parse query URL */
- if (http_parser_parse_url(RSPAMD_FSTRING_DATA(msg->url),
- RSPAMD_FSTRING_LEN(msg->url), 0, &u) != 0) {
- msg_err_session("bad request url: %V", msg->url);
-
- return FALSE;
- }
-
- if (u.field_set & (1 << UF_QUERY)) {
- /* In case if we have a query, we need to store it somewhere */
- query_args = rspamd_http_message_parse_query(msg);
- srch.begin = "File";
- srch.len = strlen("File");
- tok = g_hash_table_lookup(query_args, &srch);
-
- if (tok) {
- file_str = rspamd_mempool_ftokdup(session->pool, tok);
- session->map = rspamd_file_xmap(file_str, PROT_READ,
- &session->map_len, TRUE);
-
- if (session->map == NULL) {
- if (session->map_len != 0) {
- msg_err_session("cannot map %s: %s", file_str,
- strerror(errno));
- g_hash_table_unref(query_args);
-
- return FALSE;
- }
- }
-
- /* We need to create a new URL with file attribute removed */
- new_url = rspamd_fstring_new_init(RSPAMD_FSTRING_DATA(msg->url),
- u.field_data[UF_QUERY].off);
- new_url = rspamd_fstring_append(new_url, "?", 1);
-
- g_hash_table_iter_init(&it, query_args);
-
- while (g_hash_table_iter_next(&it, &k, &v)) {
- key_tok = k;
- tok = v;
-
- if (!rspamd_ftok_icase_equal(key_tok, &srch)) {
- rspamd_printf_fstring(&new_url, "%T=%T&",
- key_tok, tok);
- }
- }
-
- /* Erase last character (might be either & or ?) */
- rspamd_fstring_erase(new_url, new_url->len - 1, 1);
-
- rspamd_fstring_free(msg->url);
- msg->url = new_url;
- session->fname = file_str;
- }
-
- g_hash_table_unref(query_args);
- }
- }
-
- return TRUE;
- }
-
- static void
- proxy_backend_mirror_error_handler(struct rspamd_http_connection *conn, GError *err)
- {
- struct rspamd_proxy_backend_connection *bk_conn = conn->ud;
- struct rspamd_proxy_session *session;
-
- session = bk_conn->s;
- msg_info_session("abnormally closing connection from backend: %s:%s, "
- "error: %e",
- bk_conn->name,
- rspamd_inet_address_to_string_pretty(
- rspamd_upstream_addr_cur(bk_conn->up)),
- err);
-
- if (err) {
- bk_conn->err = rspamd_mempool_strdup(session->pool, err->message);
- }
-
- rspamd_upstream_fail(bk_conn->up, FALSE, err ? err->message : "unknown");
-
- proxy_backend_close_connection(bk_conn);
- REF_RELEASE(bk_conn->s);
- }
-
- static gint
- proxy_backend_mirror_finish_handler(struct rspamd_http_connection *conn,
- struct rspamd_http_message *msg)
- {
- struct rspamd_proxy_backend_connection *bk_conn = conn->ud;
- struct rspamd_proxy_session *session;
- const rspamd_ftok_t *orig_ct;
-
- session = bk_conn->s;
-
- proxy_request_decompress(msg);
- orig_ct = rspamd_http_message_find_header(msg, "Content-Type");
-
- if (!proxy_backend_parse_results(session, bk_conn, session->ctx->lua_state,
- bk_conn->parser_from_ref, msg, NULL, orig_ct)) {
- msg_warn_session("cannot parse results from the mirror backend %s:%s",
- bk_conn->name,
- rspamd_inet_address_to_string(
- rspamd_upstream_addr_cur(bk_conn->up)));
- bk_conn->err = "cannot parse ucl";
- }
-
- msg_info_session("finished mirror connection to %s", bk_conn->name);
- rspamd_upstream_ok(bk_conn->up);
-
- proxy_backend_close_connection(bk_conn);
- REF_RELEASE(bk_conn->s);
-
- return 0;
- }
-
- static void
- proxy_open_mirror_connections(struct rspamd_proxy_session *session)
- {
- gdouble coin;
- struct rspamd_http_mirror *m;
- guint i;
- struct rspamd_proxy_backend_connection *bk_conn;
- struct rspamd_http_message *msg;
- GError *err = NULL;
-
- coin = rspamd_random_double();
-
- for (i = 0; i < session->ctx->mirrors->len; i++) {
- m = g_ptr_array_index(session->ctx->mirrors, i);
-
- if (m->prob < coin) {
- /* No luck */
- continue;
- }
-
- bk_conn = rspamd_mempool_alloc0(session->pool,
- sizeof(*bk_conn));
- bk_conn->s = session;
- bk_conn->name = m->name;
- bk_conn->timeout = m->timeout;
-
- bk_conn->up = rspamd_upstream_get(m->u,
- RSPAMD_UPSTREAM_ROUND_ROBIN, NULL, 0);
- bk_conn->parser_from_ref = m->parser_from_ref;
- bk_conn->parser_to_ref = m->parser_to_ref;
-
- if (bk_conn->up == NULL) {
- msg_err_session("cannot select upstream for %s", m->name);
- continue;
- }
-
- bk_conn->backend_sock = rspamd_inet_address_connect(
- rspamd_upstream_addr_next(bk_conn->up),
- SOCK_STREAM, TRUE);
-
- if (bk_conn->backend_sock == -1) {
- msg_err_session("cannot connect upstream for %s", m->name);
- rspamd_upstream_fail(bk_conn->up, TRUE, strerror(errno));
- continue;
- }
-
- msg = rspamd_http_connection_copy_msg(session->client_message, &err);
-
- if (msg == NULL) {
- msg_err_session("cannot copy message to send to a mirror %s: %e",
- m->name, err);
- if (err) {
- g_error_free(err);
- }
- continue;
- }
-
- if (msg->url->len == 0) {
- msg->url = rspamd_fstring_append(msg->url, "/check", strlen("/check"));
- }
-
- if (m->settings_id != NULL) {
- rspamd_http_message_remove_header(msg, "Settings-ID");
- rspamd_http_message_add_header(msg, "Settings-ID", m->settings_id);
- }
-
- bk_conn->backend_conn = rspamd_http_connection_new_client_socket(
- session->ctx->http_ctx,
- NULL,
- proxy_backend_mirror_error_handler,
- proxy_backend_mirror_finish_handler,
- RSPAMD_HTTP_CLIENT_SIMPLE,
- bk_conn->backend_sock);
-
- if (m->key) {
- msg->peer_key = rspamd_pubkey_ref(m->key);
- }
-
- if (m->local ||
- rspamd_inet_address_is_local(rspamd_upstream_addr_cur(bk_conn->up))) {
-
- if (session->fname) {
- rspamd_http_message_add_header(msg, "File", session->fname);
- }
-
- msg->method = HTTP_GET;
- rspamd_http_connection_write_message_shared(bk_conn->backend_conn,
- msg, rspamd_upstream_name(bk_conn->up), NULL, bk_conn,
- bk_conn->timeout);
- }
- else {
- if (session->fname) {
- msg->flags &= ~RSPAMD_HTTP_FLAG_SHMEM;
- rspamd_http_message_set_body(msg, session->map, session->map_len);
- }
-
- msg->method = HTTP_POST;
-
- if (m->compress) {
- proxy_request_compress(msg);
-
- if (session->client_milter_conn) {
- rspamd_http_message_add_header(msg, "Content-Type",
- "application/octet-stream");
- }
- }
- else {
- if (session->client_milter_conn) {
- rspamd_http_message_add_header(msg, "Content-Type",
- "text/plain");
- }
- }
-
- rspamd_http_connection_write_message(bk_conn->backend_conn,
- msg, rspamd_upstream_name(bk_conn->up), NULL, bk_conn,
- bk_conn->timeout);
- }
-
- g_ptr_array_add(session->mirror_conns, bk_conn);
- REF_RETAIN(session);
- msg_info_session("send request to %s", m->name);
- }
- }
-
- static void
- proxy_client_write_error(struct rspamd_proxy_session *session, gint code,
- const gchar *status)
- {
- struct rspamd_http_message *reply;
-
- if (session->client_milter_conn) {
- rspamd_milter_send_action(session->client_milter_conn,
- RSPAMD_MILTER_TEMPFAIL);
- REF_RELEASE(session);
- }
- else {
- reply = rspamd_http_new_message(HTTP_RESPONSE);
-
- switch (code) {
- case ETIMEDOUT:
- reply->code = 504;
- reply->status = RSPAMD_FSTRING_LIT("Gateway timeout");
- break;
- case ECONNRESET:
- case ECONNABORTED:
- reply->code = 502;
- reply->status = RSPAMD_FSTRING_LIT("Gateway connection reset");
- break;
- case ECONNREFUSED:
- reply->code = 502;
- reply->status = RSPAMD_FSTRING_LIT("Gateway connection refused");
- break;
- default:
- if (code >= 300) {
- /* Likely HTTP error */
- reply->code = code;
- reply->status = rspamd_fstring_new_init(status, strlen(status));
- }
- else {
- reply->code = 502;
- reply->status = RSPAMD_FSTRING_LIT("Unknown gateway error: ");
- reply->status = rspamd_fstring_append(reply->status,
- status, strlen(status));
- }
- break;
- }
-
- rspamd_http_connection_write_message(session->client_conn,
- reply, NULL, NULL, session,
- session->ctx->timeout);
- }
- }
-
- static void
- proxy_backend_master_error_handler(struct rspamd_http_connection *conn, GError *err)
- {
- struct rspamd_proxy_backend_connection *bk_conn = conn->ud;
- struct rspamd_proxy_session *session;
-
- session = bk_conn->s;
- session->retries++;
- msg_info_session("abnormally closing connection from backend: %s, error: %e,"
- " retries left: %d",
- rspamd_inet_address_to_string_pretty(
- rspamd_upstream_addr_cur(session->master_conn->up)),
- err,
- session->ctx->max_retries - session->retries);
- rspamd_upstream_fail(bk_conn->up, FALSE, err ? err->message : "unknown");
- proxy_backend_close_connection(session->master_conn);
-
- if (session->ctx->max_retries > 0 &&
- session->retries >= session->ctx->max_retries) {
- msg_err_session("cannot connect to upstream, maximum retries "
- "has been reached: %d",
- session->retries);
- /* Terminate session immediately */
- if (err) {
- proxy_client_write_error(session, err->code, err->message);
- }
- else {
- proxy_client_write_error(session, 503, "Unknown error after no retries left");
- }
- }
- else {
- if (!proxy_send_master_message(session)) {
- if (err) {
- proxy_client_write_error(session, err->code, err->message);
- }
- else {
- proxy_client_write_error(session, 503, "Unknown error on write");
- }
- }
- else {
- msg_info_session("retry connection to: %s"
- " retries left: %d",
- rspamd_inet_address_to_string(
- rspamd_upstream_addr_cur(session->master_conn->up)),
- session->ctx->max_retries - session->retries);
- }
- }
- }
-
- static gint
- proxy_backend_master_finish_handler(struct rspamd_http_connection *conn,
- struct rspamd_http_message *msg)
- {
- struct rspamd_proxy_backend_connection *bk_conn = conn->ud;
- struct rspamd_proxy_session *session, *nsession;
- rspamd_fstring_t *reply;
- const rspamd_ftok_t *orig_ct;
- goffset body_offset = -1;
-
- session = bk_conn->s;
- rspamd_http_connection_steal_msg(session->master_conn->backend_conn);
- proxy_request_decompress(msg);
-
- /*
- * These are likely set by an http library, so we will double these headers
- * if they are not removed
- */
- rspamd_http_message_remove_header(msg, "Content-Length");
- rspamd_http_message_remove_header(msg, "Connection");
- rspamd_http_message_remove_header(msg, "Date");
- rspamd_http_message_remove_header(msg, "Server");
- rspamd_http_message_remove_header(msg, "Key");
- orig_ct = rspamd_http_message_find_header(msg, "Content-Type");
- rspamd_http_connection_reset(session->master_conn->backend_conn);
-
- if (!proxy_backend_parse_results(session, bk_conn, session->ctx->lua_state,
- bk_conn->parser_from_ref, msg, &body_offset, orig_ct)) {
- msg_warn_session("cannot parse results from the master backend");
- }
-
-
- if (session->legacy_support > LEGACY_SUPPORT_NO) {
- /* We need to reformat ucl to fit with legacy spamc protocol */
- if (bk_conn->results) {
- reply = rspamd_fstring_new();
-
- if (session->legacy_support == LEGACY_SUPPORT_SPAMC) {
- rspamd_ucl_tospamc_output(bk_conn->results, &reply);
- msg->flags |= RSPAMD_HTTP_FLAG_SPAMC;
- }
- else {
- rspamd_ucl_torspamc_output(bk_conn->results, &reply);
- }
-
- rspamd_http_message_set_body_from_fstring_steal(msg, reply);
- msg->method = HTTP_SYMBOLS;
- }
- else {
- msg_warn_session("cannot parse results from the master backend, "
- "return them as is");
- }
- }
-
- rspamd_upstream_ok(bk_conn->up);
-
- if (session->client_milter_conn) {
- nsession = proxy_session_refresh(session);
-
- if (body_offset > 0) {
- rspamd_milter_send_task_results(nsession->client_milter_conn,
- session->master_conn->results,
- msg->body_buf.begin + body_offset,
- msg->body_buf.len - body_offset);
- }
- else {
- rspamd_milter_send_task_results(nsession->client_milter_conn,
- session->master_conn->results, NULL, 0);
- }
- REF_RELEASE(session);
- rspamd_http_message_free(msg);
- }
- else {
- const gchar *passed_ct = NULL;
-
- if (orig_ct) {
- passed_ct = rspamd_mempool_ftokdup(session->pool, orig_ct);
- /* Remove original */
- rspamd_http_message_remove_header(msg, "Content-Type");
- }
-
- rspamd_http_connection_write_message(session->client_conn,
- msg, NULL, passed_ct, session,
- bk_conn->timeout);
- }
-
- return 0;
- }
-
- static void
- rspamd_proxy_scan_self_reply(struct rspamd_task *task)
- {
- struct rspamd_http_message *msg;
- struct rspamd_proxy_session *session = task->fin_arg, *nsession;
- ucl_object_t *rep = NULL;
- const char *ctype = "application/json";
-
- msg = rspamd_http_new_message(HTTP_RESPONSE);
- msg->date = time(NULL);
- msg->code = 200;
-
- switch (task->cmd) {
- case CMD_CHECK:
- case CMD_SKIP:
- case CMD_CHECK_RSPAMC:
- case CMD_CHECK_SPAMC:
- case CMD_CHECK_V2:
- rspamd_task_set_finish_time(task);
- rspamd_protocol_http_reply(msg, task, &rep);
- rspamd_protocol_write_log_pipe(task);
- break;
- case CMD_PING:
- rspamd_http_message_set_body(msg, "pong" CRLF, 6);
- ctype = "text/plain";
- break;
- default:
- msg_err_task("BROKEN");
- break;
- }
-
- session->master_conn->flags |= RSPAMD_BACKEND_CLOSED;
-
- if (rep) {
- session->master_conn->results = ucl_object_ref(rep);
- }
-
- if (session->client_milter_conn) {
- nsession = proxy_session_refresh(session);
-
- if (task->flags & RSPAMD_TASK_FLAG_MESSAGE_REWRITE) {
- const gchar *start;
- goffset len, hdr_off;
-
- start = task->msg.begin;
- len = task->msg.len;
-
- hdr_off = MESSAGE_FIELD(task, raw_headers_content).len;
-
- if (hdr_off < len) {
- start += hdr_off;
- len -= hdr_off;
-
- /* The problem here is that we need not end of headers, we need
- * start of body.
- *
- * Hence, we need to skip one \r\n till there is anything else in
- * a line.
- */
-
- if (*start == '\r' && len > 0) {
- start++;
- len--;
- }
-
- if (*start == '\n' && len > 0) {
- start++;
- len--;
- }
-
- rspamd_milter_send_task_results(nsession->client_milter_conn,
- session->master_conn->results, start, len);
- }
- else {
- /* XXX: should never happen! */
- rspamd_milter_send_task_results(nsession->client_milter_conn,
- session->master_conn->results, NULL, 0);
- }
- }
- else {
- rspamd_milter_send_task_results(nsession->client_milter_conn,
- session->master_conn->results, NULL, 0);
- }
- rspamd_http_message_free(msg);
- REF_RELEASE(session);
- }
- else {
- rspamd_http_connection_reset(session->client_conn);
- rspamd_http_connection_write_message(session->client_conn,
- msg,
- NULL,
- ctype,
- session,
- session->ctx->timeout / 10.0);
- }
- }
-
- static gboolean
- rspamd_proxy_task_fin(void *ud)
- {
- struct rspamd_task *task = ud;
-
- msg_debug_task("finish task");
-
- if (RSPAMD_TASK_IS_PROCESSED(task)) {
- rspamd_proxy_scan_self_reply(task);
- return TRUE;
- }
-
- if (!rspamd_task_process(task, RSPAMD_TASK_PROCESS_ALL)) {
- rspamd_proxy_scan_self_reply(task);
- return TRUE;
- }
-
- if (RSPAMD_TASK_IS_PROCESSED(task)) {
- rspamd_proxy_scan_self_reply(task);
- return TRUE;
- }
-
- /* One more iteration */
- return FALSE;
- }
-
- static gboolean
- rspamd_proxy_self_scan(struct rspamd_proxy_session *session)
- {
- struct rspamd_task *task;
- struct rspamd_http_message *msg;
- const gchar *data;
- gsize len;
-
- msg = session->client_message;
- task = rspamd_task_new(session->worker, session->ctx->cfg,
- session->pool, session->ctx->lang_det,
- session->ctx->event_loop, FALSE);
- task->flags |= RSPAMD_TASK_FLAG_MIME;
-
- if (session->ctx->milter) {
- task->protocol_flags |= RSPAMD_TASK_PROTOCOL_FLAG_MILTER |
- RSPAMD_TASK_PROTOCOL_FLAG_BODY_BLOCK;
- }
-
- task->sock = -1;
-
- if (session->client_milter_conn) {
- task->client_addr = rspamd_inet_address_copy(
- session->client_milter_conn->addr, NULL);
- }
- else {
- task->client_addr = rspamd_inet_address_copy(session->client_addr, NULL);
- }
-
- task->fin_arg = session;
- task->resolver = session->ctx->resolver;
- /* TODO: allow to disable autolearn in protocol */
- task->flags |= RSPAMD_TASK_FLAG_LEARN_AUTO;
- task->s = rspamd_session_create(task->task_pool, rspamd_proxy_task_fin,
- NULL, (event_finalizer_t) rspamd_task_free, task);
- data = rspamd_http_message_get_body(msg, &len);
-
- if (session->backend->settings_id) {
- rspamd_http_message_remove_header(msg, "Settings-ID");
- rspamd_http_message_add_header(msg, "Settings-ID",
- session->backend->settings_id);
- }
-
- /* Process message */
- if (!rspamd_protocol_handle_request(task, msg)) {
- msg_err_task("cannot handle request: %e", task->err);
- task->flags |= RSPAMD_TASK_FLAG_SKIP;
- }
- else {
- if (task->cmd == CMD_PING) {
- task->flags |= RSPAMD_TASK_FLAG_SKIP;
- }
- else {
- if (!rspamd_task_load_message(task, msg, data, len)) {
- msg_err_task("cannot load message: %e", task->err);
- task->flags |= RSPAMD_TASK_FLAG_SKIP;
- }
- }
- }
-
- /* Set global timeout for the task */
- if (session->ctx->default_upstream->timeout > 0.0) {
- task->timeout_ev.data = task;
- ev_timer_init(&task->timeout_ev, rspamd_task_timeout,
- session->ctx->default_upstream->timeout,
- session->ctx->default_upstream->timeout);
- ev_timer_start(task->event_loop, &task->timeout_ev);
- }
- else if (session->ctx->has_self_scan) {
- if (!isnan(session->ctx->task_timeout) && session->ctx->task_timeout > 0) {
- task->timeout_ev.data = task;
- ev_timer_init(&task->timeout_ev, rspamd_task_timeout,
- session->ctx->cfg->task_timeout,
- session->ctx->default_upstream->timeout);
- ev_timer_start(task->event_loop, &task->timeout_ev);
- }
- }
-
- session->master_conn->task = task;
- rspamd_task_process(task, RSPAMD_TASK_PROCESS_ALL);
-
- rspamd_session_pending(task->s);
-
- return TRUE;
- }
-
- static gboolean
- proxy_send_master_message(struct rspamd_proxy_session *session)
- {
- struct rspamd_http_message *msg;
- struct rspamd_http_upstream *backend = NULL;
- const rspamd_ftok_t *host;
- GError *err = NULL;
- gchar hostbuf[512];
-
- host = rspamd_http_message_find_header(session->client_message, "Host");
-
- if (host == NULL) {
- backend = session->ctx->default_upstream;
- }
- else {
- rspamd_strlcpy(hostbuf, host->begin, MIN(host->len + 1, sizeof(hostbuf)));
- backend = g_hash_table_lookup(session->ctx->upstreams, hostbuf);
-
- if (backend == NULL) {
- backend = session->ctx->default_upstream;
- }
- }
-
- if (backend == NULL) {
- /* No backend */
- msg_err_session("cannot find upstream for %s", host ? hostbuf : "default");
- goto err;
- }
- else {
- session->backend = backend;
-
- if (backend->self_scan) {
- return rspamd_proxy_self_scan(session);
- }
- retry:
- if (session->ctx->max_retries &&
- session->retries > session->ctx->max_retries) {
- msg_err_session("cannot connect to upstream, maximum retries "
- "has been reached: %d",
- session->retries);
- goto err;
- }
-
- /* Provide hash key if hashing based on source address is desired */
- guint hash_len;
- gpointer hash_key = rspamd_inet_address_get_hash_key(session->client_addr,
- &hash_len);
-
- if (session->ctx->max_retries > 1 &&
- session->retries == session->ctx->max_retries) {
-
- session->master_conn->up = rspamd_upstream_get_except(backend->u,
- session->master_conn->up,
- RSPAMD_UPSTREAM_ROUND_ROBIN,
- hash_key, hash_len);
- }
- else {
- session->master_conn->up = rspamd_upstream_get(backend->u,
- RSPAMD_UPSTREAM_ROUND_ROBIN,
- hash_key, hash_len);
- }
-
- session->master_conn->timeout = backend->timeout;
-
- if (session->master_conn->up == NULL) {
- msg_err_session("cannot select upstream for %s",
- host ? hostbuf : "default");
- goto err;
- }
-
- session->master_conn->backend_sock = rspamd_inet_address_connect(
- rspamd_upstream_addr_next(session->master_conn->up),
- SOCK_STREAM, TRUE);
-
- if (session->master_conn->backend_sock == -1) {
- msg_err_session("cannot connect upstream: %s(%s)",
- host ? hostbuf : "default",
- rspamd_inet_address_to_string_pretty(
- rspamd_upstream_addr_cur(
- session->master_conn->up)));
- rspamd_upstream_fail(session->master_conn->up, TRUE,
- strerror(errno));
- session->retries++;
- goto retry;
- }
-
- msg = rspamd_http_connection_copy_msg(session->client_message, &err);
- if (msg == NULL) {
- msg_err_session("cannot copy message to send it to the upstream: %e",
- err);
-
- if (err) {
- g_error_free(err);
- }
-
- goto err; /* No fallback here */
- }
-
- session->master_conn->backend_conn = rspamd_http_connection_new_client_socket(
- session->ctx->http_ctx,
- NULL,
- proxy_backend_master_error_handler,
- proxy_backend_master_finish_handler,
- RSPAMD_HTTP_CLIENT_SIMPLE,
- session->master_conn->backend_sock);
- session->master_conn->flags &= ~RSPAMD_BACKEND_CLOSED;
- session->master_conn->parser_from_ref = backend->parser_from_ref;
- session->master_conn->parser_to_ref = backend->parser_to_ref;
-
- if (backend->key) {
- msg->peer_key = rspamd_pubkey_ref(backend->key);
- }
-
- if (backend->settings_id != NULL) {
- rspamd_http_message_remove_header(msg, "Settings-ID");
- rspamd_http_message_add_header(msg, "Settings-ID",
- backend->settings_id);
- }
-
- if (backend->local ||
- rspamd_inet_address_is_local(
- rspamd_upstream_addr_cur(
- session->master_conn->up))) {
-
- if (session->fname) {
- rspamd_http_message_add_header(msg, "File", session->fname);
- }
-
- msg->method = HTTP_GET;
-
- rspamd_http_connection_write_message_shared(
- session->master_conn->backend_conn,
- msg, rspamd_upstream_name(session->master_conn->up),
- NULL, session->master_conn,
- session->master_conn->timeout);
- }
- else {
- if (session->fname) {
- msg->flags &= ~RSPAMD_HTTP_FLAG_SHMEM;
- rspamd_http_message_set_body(msg,
- session->map, session->map_len);
- }
-
- msg->method = HTTP_POST;
-
- if (backend->compress) {
- proxy_request_compress(msg);
- if (session->client_milter_conn) {
- rspamd_http_message_add_header(msg, "Content-Type",
- "application/octet-stream");
- }
- }
- else {
- if (session->client_milter_conn) {
- rspamd_http_message_add_header(msg, "Content-Type",
- "text/plain");
- }
- }
-
- rspamd_http_connection_write_message(
- session->master_conn->backend_conn,
- msg, rspamd_upstream_name(session->master_conn->up),
- NULL, session->master_conn,
- session->master_conn->timeout);
- }
- }
-
- return TRUE;
-
- err:
- if (session->client_milter_conn) {
- rspamd_milter_send_action(session->client_milter_conn,
- RSPAMD_MILTER_TEMPFAIL);
- REF_RELEASE(session);
- }
- else {
- rspamd_http_connection_steal_msg(session->client_conn);
- rspamd_http_connection_reset(session->client_conn);
- proxy_client_write_error(session, 404, "Backend not found");
- }
-
- return FALSE;
- }
-
- static void
- proxy_client_error_handler(struct rspamd_http_connection *conn, GError *err)
- {
- struct rspamd_proxy_session *session = conn->ud;
-
- msg_info_session("abnormally closing connection from: %s, error: %s",
- rspamd_inet_address_to_string(session->client_addr), err->message);
- /* Terminate session immediately */
- proxy_backend_close_connection(session->master_conn);
- REF_RELEASE(session);
- }
-
- static gint
- proxy_client_finish_handler(struct rspamd_http_connection *conn,
- struct rspamd_http_message *msg)
- {
- struct rspamd_proxy_session *session = conn->ud;
-
- if (!session->master_conn) {
- session->master_conn = rspamd_mempool_alloc0(session->pool,
- sizeof(*session->master_conn));
- session->master_conn->s = session;
- session->master_conn->name = "master";
-
- /* Reset spamc legacy */
- if (msg->method >= HTTP_SYMBOLS) {
- msg->method = HTTP_POST;
-
- if (msg->flags & RSPAMD_HTTP_FLAG_SPAMC) {
- session->legacy_support = LEGACY_SUPPORT_SPAMC;
- msg_info_session("enabling legacy spamc mode for session");
- }
- else {
- session->legacy_support = LEGACY_SUPPORT_RSPAMC;
- msg_info_session("enabling legacy rspamc mode for session");
- }
- }
-
- if (msg->url->len == 0) {
- msg->url = rspamd_fstring_append(msg->url,
- "/" MSG_CMD_CHECK_V2, strlen("/" MSG_CMD_CHECK_V2));
- }
-
- if (!proxy_check_file(msg, session)) {
- goto err;
- }
-
- session->client_message = rspamd_http_connection_steal_msg(
- session->client_conn);
- session->shmem_ref = rspamd_http_message_shmem_ref(session->client_message);
- rspamd_http_message_remove_header(msg, "Content-Length");
- rspamd_http_message_remove_header(msg, "Transfer-Encoding");
- rspamd_http_message_remove_header(msg, "Keep-Alive");
- rspamd_http_message_remove_header(msg, "Connection");
- rspamd_http_message_remove_header(msg, "Key");
-
- proxy_open_mirror_connections(session);
- rspamd_http_connection_reset(session->client_conn);
-
- proxy_send_master_message(session);
- }
- else {
- msg_info_session("finished master connection");
- proxy_backend_close_connection(session->master_conn);
- REF_RELEASE(session);
- }
-
- return 0;
-
- err:
- rspamd_http_connection_steal_msg(session->client_conn);
- rspamd_http_message_remove_header(msg, "Content-Length");
- rspamd_http_message_remove_header(msg, "Key");
- rspamd_http_message_remove_header(msg, "Transfer-Encoding");
- rspamd_http_message_remove_header(msg, "Keep-Alive");
- rspamd_http_message_remove_header(msg, "Connection");
- rspamd_http_connection_reset(session->client_conn);
- proxy_client_write_error(session, 404, "Backend not found");
-
- return 0;
- }
-
- static void
- proxy_milter_finish_handler(gint fd,
- struct rspamd_milter_session *rms,
- void *ud)
- {
- struct rspamd_proxy_session *session = ud;
- struct rspamd_http_message *msg;
-
- session->client_milter_conn = rms;
-
- if (rms->message == NULL || rms->message->len == 0) {
- msg_info_session("finished milter connection");
- proxy_backend_close_connection(session->master_conn);
- REF_RELEASE(session);
- }
- else {
- if (!session->master_conn) {
- session->master_conn = rspamd_mempool_alloc0(session->pool,
- sizeof(*session->master_conn));
- }
-
- msg = rspamd_milter_to_http(rms);
- session->master_conn->s = session;
- session->master_conn->name = "master";
- session->client_message = msg;
-
- proxy_open_mirror_connections(session);
- proxy_send_master_message(session);
- }
- }
-
- static void
- proxy_milter_error_handler(gint fd,
- struct rspamd_milter_session *rms, /* unused */
- void *ud, GError *err)
- {
- struct rspamd_proxy_session *session = ud;
-
- if (err && err->code != 0) {
- msg_info_session("abnormally closing milter connection from: %s, "
- "error: %e",
- rspamd_inet_address_to_string_pretty(session->client_addr),
- err);
- /* Terminate session immediately */
- proxy_backend_close_connection(session->master_conn);
- REF_RELEASE(session);
- }
- else {
- msg_info_session("normally closing milter connection from: %s, "
- "%e",
- rspamd_inet_address_to_string_pretty(session->client_addr),
- err);
- /* Terminate session immediately */
- proxy_backend_close_connection(session->master_conn);
- REF_RELEASE(session);
- }
- }
-
- static void
- proxy_accept_socket(EV_P_ ev_io *w, int revents)
- {
- struct rspamd_worker *worker = (struct rspamd_worker *) w->data;
- struct rspamd_proxy_ctx *ctx;
- rspamd_inet_addr_t *addr = NULL;
- struct rspamd_proxy_session *session;
- gint nfd;
-
- ctx = worker->ctx;
-
- if ((nfd =
- rspamd_accept_from_socket(w->fd, &addr,
- rspamd_worker_throttle_accept_events, worker->accept_events)) == -1) {
- msg_warn("accept failed: %s", strerror(errno));
- return;
- }
- /* Check for EAGAIN */
- if (nfd == 0) {
- rspamd_inet_address_free(addr);
- return;
- }
-
- session = g_malloc0(sizeof(*session));
- REF_INIT_RETAIN(session, proxy_session_dtor);
- session->client_sock = nfd;
- session->client_addr = addr;
- session->mirror_conns = g_ptr_array_sized_new(ctx->mirrors->len);
-
- session->pool = rspamd_mempool_new(rspamd_mempool_suggest_size(),
- "proxy", 0);
- session->ctx = ctx;
- session->worker = worker;
-
- if (ctx->sessions_cache) {
- rspamd_worker_session_cache_add(ctx->sessions_cache,
- session->pool->tag.uid, &session->ref.refcount, session);
- }
-
- if (!ctx->milter) {
- session->client_conn = rspamd_http_connection_new_server(
- ctx->http_ctx,
- nfd,
- NULL,
- proxy_client_error_handler,
- proxy_client_finish_handler,
- 0);
-
- if (ctx->key) {
- rspamd_http_connection_set_key(session->client_conn, ctx->key);
- }
-
- msg_info_session("accepted http connection from %s port %d",
- rspamd_inet_address_to_string(addr),
- rspamd_inet_address_get_port(addr));
-
- rspamd_http_connection_read_message_shared(session->client_conn,
- session,
- session->ctx->timeout);
- }
- else {
- msg_info_session("accepted milter connection from %s port %d",
- rspamd_inet_address_to_string(addr),
- rspamd_inet_address_get_port(addr));
-
- #ifdef TCP_NODELAY
-
- #ifndef SOL_TCP
- #define SOL_TCP IPPROTO_TCP
- #endif
-
- if (rspamd_inet_address_get_af(addr) != AF_UNIX) {
- gint sopt = 1;
-
- if (setsockopt(nfd, SOL_TCP, TCP_NODELAY, &sopt, sizeof(sopt)) ==
- -1) {
- msg_warn_session("cannot set TCP_NODELAY: %s",
- strerror(errno));
- }
- }
- #endif
-
- rspamd_milter_handle_socket(nfd, 0.0,
- session->pool,
- ctx->event_loop,
- proxy_milter_finish_handler,
- proxy_milter_error_handler,
- session);
- }
- }
-
- static void
- adjust_upstreams_limits(struct rspamd_proxy_ctx *ctx)
- {
- struct rspamd_http_upstream *backend;
- gpointer k, v;
- GHashTableIter it;
-
- /*
- * We set error time equal to max_retries * backend_timeout and max_errors
- * to max_retries - 1
- *
- * So if we failed to scan a message on a backend for some reasons, we
- * will try to re-resolve it faster
- */
-
- g_hash_table_iter_init(&it, ctx->upstreams);
-
- while (g_hash_table_iter_next(&it, &k, &v)) {
- backend = (struct rspamd_http_upstream *) v;
-
- if (!backend->self_scan && backend->u) {
- rspamd_upstreams_set_limits(backend->u,
- NAN, NAN, ctx->max_retries * backend->timeout, NAN,
- ctx->max_retries - 1, 0);
- }
- }
- }
-
- __attribute__((noreturn)) void
- start_rspamd_proxy(struct rspamd_worker *worker)
- {
- struct rspamd_proxy_ctx *ctx = worker->ctx;
- gboolean is_controller = FALSE;
-
- g_assert(rspamd_worker_check_context(worker->ctx, rspamd_rspamd_proxy_magic));
- ctx->cfg = worker->srv->cfg;
- ctx->event_loop = rspamd_prepare_worker(worker, "rspamd_proxy",
- proxy_accept_socket);
-
- 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);
- rspamd_mempool_add_destructor(ctx->cfg->cfg_pool,
- (rspamd_mempool_destruct_t) rspamd_http_context_free,
- ctx->http_ctx);
-
- if (ctx->has_self_scan) {
- /* Additional initialisation needed */
- rspamd_worker_init_scanner(worker, ctx->event_loop, ctx->resolver,
- &ctx->lang_det);
- ctx->task_timeout = rspamd_worker_check_and_adjust_timeout(ctx->cfg, NAN);
-
- is_controller = rspamd_worker_check_controller_presence(worker);
- }
- else {
- worker->flags &= ~RSPAMD_WORKER_SCANNER;
- }
-
- if (worker->srv->cfg->enable_sessions_cache) {
- ctx->sessions_cache = rspamd_worker_session_cache_new(worker,
- ctx->event_loop);
- }
-
- ctx->milter_ctx.spam_header = ctx->spam_header;
- ctx->milter_ctx.discard_on_reject = ctx->discard_on_reject;
- ctx->milter_ctx.quarantine_on_reject = ctx->quarantine_on_reject;
- ctx->milter_ctx.sessions_cache = ctx->sessions_cache;
- ctx->milter_ctx.client_ca_name = ctx->client_ca_name;
- ctx->milter_ctx.reject_message = ctx->reject_message;
- ctx->milter_ctx.cfg = ctx->cfg;
- rspamd_milter_init_library(&ctx->milter_ctx);
-
- if (is_controller) {
- rspamd_worker_init_controller(worker, NULL);
- }
- else {
- if (ctx->has_self_scan) {
- rspamd_map_watch(worker->srv->cfg, ctx->event_loop, ctx->resolver,
- worker, RSPAMD_MAP_WATCH_SCANNER);
- }
- else {
- rspamd_map_watch(worker->srv->cfg, ctx->event_loop, ctx->resolver,
- worker, RSPAMD_MAP_WATCH_WORKER);
- }
- }
-
- rspamd_lua_run_postloads(ctx->cfg->lua_state, ctx->cfg, ctx->event_loop,
- worker);
- adjust_upstreams_limits(ctx);
-
- ev_loop(ctx->event_loop, 0);
- rspamd_worker_block_signals();
-
- if (ctx->has_self_scan) {
- rspamd_stat_close();
- }
-
- if (is_controller) {
- rspamd_controller_on_terminate(worker, NULL);
- }
-
- REF_RELEASE(ctx->cfg);
- rspamd_log_close(worker->srv->logger);
- rspamd_unset_crash_handler(worker->srv);
-
- exit(EXIT_SUCCESS);
- }
|