From 4301ec16d04b9319088f6716a32ef416fc4ab43d Mon Sep 17 00:00:00 2001 From: mdecimus Date: Tue, 12 Dec 2023 15:28:21 +0100 Subject: [PATCH] Key/value store access from Sieve scripts --- Cargo.lock | 185 +++++++-------- crates/jmap/src/api/http.rs | 2 +- crates/jmap/src/lib.rs | 8 +- crates/jmap/src/services/housekeeper.rs | 72 ++---- crates/main/src/main.rs | 15 +- crates/nlp/src/bayes/mod.rs | 15 ++ crates/smtp/Cargo.toml | 3 +- crates/smtp/src/config/scripts.rs | 34 ++- crates/smtp/src/core/mod.rs | 5 +- crates/smtp/src/scripts/event_loop.rs | 12 +- crates/smtp/src/scripts/plugins/bayes.rs | 194 ++++++++-------- crates/smtp/src/scripts/plugins/lookup.rs | 218 ++++++++++++++---- crates/smtp/src/scripts/plugins/mod.rs | 10 +- crates/smtp/src/scripts/plugins/query.rs | 2 +- crates/store/src/config.rs | 73 +++++- crates/store/src/dispatch/blob.rs | 88 +++++++ crates/store/src/dispatch/fts.rs | 82 +++++++ crates/store/src/dispatch/lookup.rs | 182 +++++++++++++++ crates/store/src/dispatch/mod.rs | 27 +++ .../src/{dispatch.rs => dispatch/store.rs} | 153 +----------- crates/store/src/fts/index.rs | 8 +- crates/store/src/lib.rs | 15 ++ crates/store/src/query/acl.rs | 4 +- crates/store/src/write/blob.rs | 17 +- crates/store/src/write/key.rs | 7 +- crates/store/src/write/mod.rs | 5 +- crates/store/src/write/purge.rs | 77 +++++++ .../spamfilter/scripts/bayes_classify.sieve | 2 +- .../config/spamfilter/scripts/config.sieve | 6 +- .../config/spamfilter/scripts/dmarc.sieve | 4 +- .../config/spamfilter/scripts/epilogue.sieve | 4 +- .../config/spamfilter/scripts/from.sieve | 8 +- .../config/spamfilter/scripts/greylist.sieve | 6 +- .../config/spamfilter/scripts/mime.sieve | 4 +- resources/config/spamfilter/scripts/rbl.sieve | 4 +- .../config/spamfilter/scripts/recipient.sieve | 4 +- .../spamfilter/scripts/replies_in.sieve | 12 +- .../spamfilter/scripts/replies_out.sieve | 6 +- .../config/spamfilter/scripts/replyto.sieve | 6 +- .../spamfilter/scripts/reputation.sieve | 15 +- .../config/spamfilter/scripts/scores.sieve | 2 +- .../config/spamfilter/scripts/spamtrap.sieve | 4 +- resources/config/spamfilter/scripts/url.sieve | 10 +- tests/src/smtp/inbound/antispam.rs | 45 ---- tests/src/smtp/mod.rs | 16 +- tests/src/store/blob.rs | 8 +- 46 files changed, 1087 insertions(+), 592 deletions(-) create mode 100644 crates/store/src/dispatch/blob.rs create mode 100644 crates/store/src/dispatch/fts.rs create mode 100644 crates/store/src/dispatch/lookup.rs create mode 100644 crates/store/src/dispatch/mod.rs rename crates/store/src/{dispatch.rs => dispatch/store.rs} (75%) create mode 100644 crates/store/src/write/purge.rs diff --git a/Cargo.lock b/Cargo.lock index dc68e572..a69bbbb2 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -125,9 +125,9 @@ dependencies = [ [[package]] name = "anstream" -version = "0.6.4" +version = "0.6.5" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "2ab91ebe16eb252986481c5b62f6098f3b698a45e34b5b98200cf20dd2484a44" +checksum = "d664a92ecae85fd0a7392615844904654d1d5f5514837f471ddef4a057aba1b6" dependencies = [ "anstyle", "anstyle-parse", @@ -154,9 +154,9 @@ dependencies = [ [[package]] name = "anstyle-query" -version = "1.0.1" +version = "1.0.2" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a3a318f1f38d2418400f8209655bfd825785afd25aa30bb7ba6cc792e4596748" +checksum = "e28923312444cdd728e4738b3f9c9cac739500909bb3d3c94b43551b16517648" dependencies = [ "windows-sys 0.52.0", ] @@ -270,7 +270,7 @@ checksum = "5fd55a5ba1179988837d24ab4c7cc8ed6efdeff578ede0416b4225a5fca35bd0" dependencies = [ "proc-macro2", "quote", - "syn 2.0.39", + "syn 2.0.40", ] [[package]] @@ -292,7 +292,7 @@ checksum = "16e62a023e7c117e27523144c5d2459f4397fcc3cab0085af8e2224f643a0193" dependencies = [ "proc-macro2", "quote", - "syn 2.0.39", + "syn 2.0.40", ] [[package]] @@ -303,7 +303,7 @@ checksum = "a66537f1bb974b254c98ed142ff995236e81b9d0fe4db0575f46612cb15eb0f9" dependencies = [ "proc-macro2", "quote", - "syn 2.0.39", + "syn 2.0.40", ] [[package]] @@ -366,7 +366,7 @@ dependencies = [ "bytes", "futures-util", "http 0.2.11", - "http-body 0.4.5", + "http-body 0.4.6", "hyper 0.14.27", "itoa", "matchit", @@ -392,7 +392,7 @@ dependencies = [ "bytes", "futures-util", "http 0.2.11", - "http-body 0.4.5", + "http-body 0.4.6", "mime", "rustversion", "tower-layer", @@ -485,7 +485,7 @@ dependencies = [ "regex", "rustc-hash", "shlex", - "syn 2.0.39", + "syn 2.0.40", "which", ] @@ -506,7 +506,7 @@ dependencies = [ "regex", "rustc-hash", "shlex", - "syn 2.0.39", + "syn 2.0.40", ] [[package]] @@ -648,7 +648,7 @@ dependencies = [ "proc-macro-crate 2.0.0", "proc-macro2", "quote", - "syn 2.0.39", + "syn 2.0.40", "syn_derive", ] @@ -903,7 +903,7 @@ dependencies = [ "heck", "proc-macro2", "quote", - "syn 2.0.39", + "syn 2.0.40", ] [[package]] @@ -1170,7 +1170,7 @@ checksum = "f46882e17999c6cc590af592290432be3bce0428cb0d5f8b6715e4dc7b383eb3" dependencies = [ "proc-macro2", "quote", - "syn 2.0.39", + "syn 2.0.40", ] [[package]] @@ -1218,7 +1218,7 @@ dependencies = [ "proc-macro2", "quote", "strsim", - "syn 2.0.39", + "syn 2.0.40", ] [[package]] @@ -1240,7 +1240,7 @@ checksum = "836a9bbc7ad63342d6d6e7b815ccab164bc77a2d95d84bc3117a8c0d5c98e2d5" dependencies = [ "darling_core 0.20.3", "quote", - "syn 2.0.39", + "syn 2.0.40", ] [[package]] @@ -1414,7 +1414,7 @@ dependencies = [ "pbkdf2 0.12.2", "pwhash", "regex", - "rustls 0.22.0", + "rustls 0.22.1", "rustls-pki-types", "scrypt", "sha1", @@ -1476,7 +1476,7 @@ checksum = "487585f4d0c6655fe74905e2504d8ad6908e4db67f744eb140876906c2f3175d" dependencies = [ "proc-macro2", "quote", - "syn 2.0.39", + "syn 2.0.40", ] [[package]] @@ -1678,7 +1678,7 @@ dependencies = [ "heck", "proc-macro2", "quote", - "syn 2.0.39", + "syn 2.0.40", ] [[package]] @@ -1717,9 +1717,9 @@ checksum = "7360491ce676a36bf9bb3c56c1aa791658183a54d2744120f27285738d90465a" [[package]] name = "fancy-regex" -version = "0.11.0" +version = "0.12.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b95f7c0680e4142284cf8b22c14a476e87d61b004a3a0861872b32ef7ead40a2" +checksum = "7493d4c459da9f84325ad297371a6b2b8a162800873a22e3b6b6512e61d18c05" dependencies = [ "bit-set", "regex", @@ -1878,7 +1878,7 @@ checksum = "83c8d52fe8b46ab822b4decdcc0d6d85aeedfc98f0d52ba2bd4aec4a97807516" dependencies = [ "proc-macro2", "quote", - "syn 2.0.39", + "syn 2.0.40", "try_map", ] @@ -1916,7 +1916,7 @@ checksum = "b0fa992f1656e1707946bbba340ad244f0814009ef8c0118eb7b658395f19a2e" dependencies = [ "frunk_proc_macro_helpers", "quote", - "syn 2.0.39", + "syn 2.0.40", ] [[package]] @@ -1928,7 +1928,7 @@ dependencies = [ "frunk_core", "proc-macro2", "quote", - "syn 2.0.39", + "syn 2.0.40", ] [[package]] @@ -1940,7 +1940,7 @@ dependencies = [ "frunk_core", "frunk_proc_macro_helpers", "quote", - "syn 2.0.39", + "syn 2.0.40", ] [[package]] @@ -2005,7 +2005,7 @@ checksum = "53b153fd91e4b0147f4aced87be237c98248656bb01050b96bf3ee89220a8ddb" dependencies = [ "proc-macro2", "quote", - "syn 2.0.39", + "syn 2.0.40", ] [[package]] @@ -2216,7 +2216,7 @@ dependencies = [ "once_cell", "rand", "ring 0.16.20", - "rustls 0.21.9", + "rustls 0.21.10", "rustls-pemfile 1.0.4", "thiserror", "tinyvec", @@ -2241,7 +2241,7 @@ dependencies = [ "parking_lot", "rand", "resolv-conf", - "rustls 0.21.9", + "rustls 0.21.10", "smallvec", "thiserror", "tokio", @@ -2321,9 +2321,9 @@ dependencies = [ [[package]] name = "http-body" -version = "0.4.5" +version = "0.4.6" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d5f38f16d184e36f2408a55281cd658ecbd3ca05cce6d6510a176eca393e26d1" +checksum = "7ceab25649e9960c0311ea418d17bee82c0dcec1bd053b5f9a66e265a693bed2" dependencies = [ "bytes", "http 0.2.11", @@ -2383,7 +2383,7 @@ dependencies = [ "futures-util", "h2 0.3.22", "http 0.2.11", - "http-body 0.4.5", + "http-body 0.4.6", "httparse", "httpdate", "itoa", @@ -2423,7 +2423,7 @@ dependencies = [ "futures-util", "http 0.2.11", "hyper 0.14.27", - "rustls 0.21.9", + "rustls 0.21.10", "tokio", "tokio-rustls 0.24.1", ] @@ -2540,7 +2540,7 @@ dependencies = [ "nlp", "parking_lot", "rand", - "rustls 0.22.0", + "rustls 0.22.1", "rustls-pemfile 2.0.0", "store", "tokio", @@ -2662,9 +2662,9 @@ dependencies = [ [[package]] name = "itoa" -version = "1.0.9" +version = "1.0.10" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "af150ab688ff2122fcef229be89cb50dd66af9e01a4ff320cc137eecc9bacc38" +checksum = "b1a46d1a171d865aa5f83f92695765caa047a9b4cbae2cbf37dbd613a793fd4c" [[package]] name = "jemalloc-sys" @@ -2767,7 +2767,7 @@ dependencies = [ "maybe-async", "parking_lot", "reqwest", - "rustls 0.21.9", + "rustls 0.21.10", "serde", "serde_json", "tokio", @@ -2905,7 +2905,7 @@ dependencies = [ "nom", "percent-encoding", "ring 0.16.20", - "rustls 0.21.9", + "rustls 0.21.10", "rustls-native-certs", "thiserror", "tokio", @@ -2918,9 +2918,9 @@ dependencies = [ [[package]] name = "libc" -version = "0.2.150" +version = "0.2.151" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "89d92a4743f9a61002fae18374ed11e7973f530cb3a3255fb354818118b2203c" +checksum = "302d7ab3130588088d277783b1e2d2e10c9e9e4a16dd9050e6ec93fb3e7048f4" [[package]] name = "libloading" @@ -3097,7 +3097,7 @@ dependencies = [ "base64 0.21.5", "gethostname", "md5", - "rustls 0.22.0", + "rustls 0.22.1", "rustls-pki-types", "smtp-proto", "tokio", @@ -3137,7 +3137,7 @@ dependencies = [ "mail-send", "md5", "parking_lot", - "rustls 0.22.0", + "rustls 0.22.1", "rustls-pemfile 2.0.0", "sieve-rs", "store", @@ -3298,7 +3298,7 @@ dependencies = [ "proc-macro-error", "proc-macro2", "quote", - "syn 2.0.39", + "syn 2.0.40", "termcolor", "thiserror", ] @@ -3325,7 +3325,7 @@ dependencies = [ "percent-encoding", "pin-project", "rand", - "rustls 0.21.9", + "rustls 0.21.10", "rustls-pemfile 1.0.4", "serde", "serde_json", @@ -3575,7 +3575,7 @@ checksum = "a948666b637a0f465e8564c73e89d4dde00d72d4d473cc972f390fc3dcee7d9c" dependencies = [ "proc-macro2", "quote", - "syn 2.0.39", + "syn 2.0.40", ] [[package]] @@ -3586,9 +3586,9 @@ checksum = "ff011a302c396a5197692431fc1948019154afc178baf7d8e37367442a4601cf" [[package]] name = "openssl-src" -version = "300.1.6+3.1.4" +version = "300.2.1+3.2.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "439fac53e092cd7442a3660c85dde4643ab3b5bd39040912388dcdabf6b88085" +checksum = "3fe476c29791a5ca0d1273c697e96085bbabbbea2ef7afd5617e78a4b40332d3" dependencies = [ "cc", ] @@ -3886,7 +3886,7 @@ dependencies = [ "phf_shared 0.11.2", "proc-macro2", "quote", - "syn 2.0.39", + "syn 2.0.40", ] [[package]] @@ -3924,7 +3924,7 @@ checksum = "4359fd9c9171ec6e8c62926d6faaf553a8dc3f64e1507e76da7911b4f6a04405" dependencies = [ "proc-macro2", "quote", - "syn 2.0.39", + "syn 2.0.40", ] [[package]] @@ -4044,7 +4044,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ae005bd773ab59b4725093fd7df83fd7892f7d8eafb48dbd7de6e024e4215f9d" dependencies = [ "proc-macro2", - "syn 2.0.39", + "syn 2.0.40", ] [[package]] @@ -4440,7 +4440,7 @@ dependencies = [ "futures-util", "h2 0.3.22", "http 0.2.11", - "http-body 0.4.5", + "http-body 0.4.6", "hyper 0.14.27", "hyper-rustls", "ipnet", @@ -4451,7 +4451,7 @@ dependencies = [ "once_cell", "percent-encoding", "pin-project-lite", - "rustls 0.21.9", + "rustls 0.21.10", "rustls-pemfile 1.0.4", "serde", "serde_json", @@ -4747,9 +4747,9 @@ dependencies = [ [[package]] name = "rustix" -version = "0.38.26" +version = "0.38.28" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "9470c4bf8246c8daf25f9598dca807fb6510347b1e1cfa55749113850c79d88a" +checksum = "72e572a5e8ca657d7366229cdde4bd14c4eb5499a9573d4d366fe1b599daa316" dependencies = [ "bitflags 2.4.1", "errno", @@ -4772,9 +4772,9 @@ dependencies = [ [[package]] name = "rustls" -version = "0.21.9" +version = "0.21.10" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "629648aced5775d558af50b2b4c7b02983a04b312126d45eeead26e7caa498b9" +checksum = "f9d5a6813c0759e4609cd494e8e725babae6a2ca7b62a5536a13daaec6fcb7ba" dependencies = [ "log", "ring 0.17.7", @@ -4784,9 +4784,9 @@ dependencies = [ [[package]] name = "rustls" -version = "0.22.0" +version = "0.22.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "5bc238b76c51bbc449c55ffbc39d03772a057cc8cf783c49d4af4c2537b74a8b" +checksum = "fe6b63262c9fcac8659abfaa96cac103d28166d3ff3eaf8f412e19f3ae9e5a48" dependencies = [ "log", "ring 0.17.7", @@ -4862,9 +4862,9 @@ checksum = "7ffc183a10b4478d04cbbbfc96d0873219d962dd5accaff2ffbd4ceb7df837f4" [[package]] name = "ryu" -version = "1.0.15" +version = "1.0.16" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "1ad4cc8da4ef723ed60bced201181d83791ad433213d8c24efffda1eec85d741" +checksum = "f98d2aa92eebf49b69786be48e4477826b256916e84a57ff2a4f21923b48eb4c" [[package]] name = "salsa20" @@ -5071,7 +5071,7 @@ checksum = "43576ca501357b9b071ac53cdc7da8ef0cbd9493d8df094cd821777ea6e894d3" dependencies = [ "proc-macro2", "quote", - "syn 2.0.39", + "syn 2.0.40", ] [[package]] @@ -5141,7 +5141,7 @@ checksum = "91d129178576168c589c9ec973feedf7d3126c01ac2bf08795109aa35b69fb8f" dependencies = [ "proc-macro2", "quote", - "syn 2.0.39", + "syn 2.0.40", ] [[package]] @@ -5227,7 +5227,7 @@ checksum = "a7cee0529a6d40f580e7a5e6c495c8fbfe21b7b52795ed4bb5e62cdf92bc6380" [[package]] name = "sieve-rs" version = "0.3.1" -source = "git+https://github.com/stalwartlabs/sieve#b7b62b3b33d05b7c1e50ea5123a35ef461c66b8d" +source = "git+https://github.com/stalwartlabs/sieve#be17314cee22f7c3f154ed2e0e5f669bb098e83b" dependencies = [ "ahash 0.8.6", "bincode", @@ -5295,6 +5295,7 @@ name = "smtp" version = "0.4.2" dependencies = [ "ahash 0.8.6", + "bincode", "blake3", "dashmap", "decancer", @@ -5320,7 +5321,7 @@ dependencies = [ "rayon", "regex", "reqwest", - "rustls 0.22.0", + "rustls 0.22.1", "rustls-pemfile 2.0.0", "rustls-pki-types", "serde", @@ -5489,7 +5490,7 @@ dependencies = [ "rocksdb", "rusqlite", "rust-s3", - "rustls 0.22.0", + "rustls 0.22.1", "rustls-pki-types", "serde", "serde_json", @@ -5560,9 +5561,9 @@ dependencies = [ [[package]] name = "syn" -version = "2.0.39" +version = "2.0.40" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "23e78b90f2fcf45d3e842032ce32e3f2d1545ba6636271dcbf24fa306d87be7a" +checksum = "13fa70a4ee923979ffb522cacce59d34421ebdea5625e1073c4326ef9d2dd42e" dependencies = [ "proc-macro2", "quote", @@ -5578,7 +5579,7 @@ dependencies = [ "proc-macro-error", "proc-macro2", "quote", - "syn 2.0.39", + "syn 2.0.40", ] [[package]] @@ -5702,7 +5703,7 @@ dependencies = [ "num_cpus", "rayon", "reqwest", - "rustls 0.22.0", + "rustls 0.22.1", "rustls-pemfile 2.0.0", "rustls-pki-types", "serde", @@ -5736,7 +5737,7 @@ checksum = "266b2e40bc00e5a6c09c3584011e08b06f123c00362c92b975ba9843aaaa14b8" dependencies = [ "proc-macro2", "quote", - "syn 2.0.39", + "syn 2.0.40", ] [[package]] @@ -5814,9 +5815,9 @@ checksum = "1f3ccbac311fea05f86f61904b462b55fb3df8837a366dfc601a0161d0532f20" [[package]] name = "tokio" -version = "1.34.0" +version = "1.35.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d0c014766411e834f7af5b8f4cf46257aab4036ca95e9d2c144a10f59ad6f5b9" +checksum = "841d45b238a16291a4e1584e61820b8ae57d696cc5015c459c229ccc6990cc1c" dependencies = [ "backtrace", "bytes", @@ -5849,7 +5850,7 @@ checksum = "5b8a1e28f2deaa14e508979454cb3a223b10b938b45af148bc0986de36f1923b" dependencies = [ "proc-macro2", "quote", - "syn 2.0.39", + "syn 2.0.40", ] [[package]] @@ -5884,7 +5885,7 @@ version = "0.24.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c28327cf380ac148141087fbfb9de9d7bd4e84ab5d2c28fbc911d753de8a7081" dependencies = [ - "rustls 0.21.9", + "rustls 0.21.10", "tokio", ] @@ -5894,7 +5895,7 @@ version = "0.25.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "775e0c0f0adb3a2f22a00c4745d728b479985fc15ee7ca6a2608388c5569860f" dependencies = [ - "rustls 0.22.0", + "rustls 0.22.1", "rustls-pki-types", "tokio", ] @@ -5918,7 +5919,7 @@ checksum = "212d5dcb2a1ce06d81107c3d0ffa3121fe974b73f068c8282cb1c32328113b6c" dependencies = [ "futures-util", "log", - "rustls 0.21.9", + "rustls 0.21.10", "tokio", "tokio-rustls 0.24.1", "tungstenite", @@ -5981,7 +5982,7 @@ dependencies = [ "futures-util", "h2 0.3.22", "http 0.2.11", - "http-body 0.4.5", + "http-body 0.4.6", "hyper 0.14.27", "hyper-timeout", "percent-encoding", @@ -6059,7 +6060,7 @@ checksum = "34704c8d6ebcbc939824180af020566b01a7c01f80641264eba0999f6c2b6be7" dependencies = [ "proc-macro2", "quote", - "syn 2.0.39", + "syn 2.0.40", ] [[package]] @@ -6155,7 +6156,7 @@ dependencies = [ "httparse", "log", "rand", - "rustls 0.21.9", + "rustls 0.21.10", "sha1", "thiserror", "url", @@ -6327,7 +6328,7 @@ dependencies = [ "opentelemetry_sdk", "privdrop", "rand", - "rustls 0.22.0", + "rustls 0.22.1", "rustls-pemfile 2.0.0", "rustls-pki-types", "serde", @@ -6411,7 +6412,7 @@ dependencies = [ "once_cell", "proc-macro2", "quote", - "syn 2.0.39", + "syn 2.0.40", "wasm-bindgen-shared", ] @@ -6445,7 +6446,7 @@ checksum = "f0eb82fcb7930ae6219a7ecfd55b217f5f0893484b7a13022ebb2b2bf20b5283" dependencies = [ "proc-macro2", "quote", - "syn 2.0.39", + "syn 2.0.40", "wasm-bindgen-backend", "wasm-bindgen-shared", ] @@ -6801,9 +6802,9 @@ checksum = "dff9641d1cd4be8d1a070daf9e3773c5f67e78b4d9d42263020c057706765c04" [[package]] name = "winnow" -version = "0.5.26" +version = "0.5.27" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b67b5f0a4e7a27a64c651977932b9dc5667ca7fc31ac44b03ed37a0cf42fdfff" +checksum = "cb877ca3232bec99a6472ed63f7241de2a250165260908b2d24c09d867907a85" dependencies = [ "memchr", ] @@ -6857,11 +6858,13 @@ dependencies = [ [[package]] name = "xattr" -version = "1.0.1" +version = "1.1.2" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f4686009f71ff3e5c4dbcf1a282d0a44db3f021ba69350cd42086b3e5f1c6985" +checksum = "d367426ae76bdfce3d8eaea6e94422afd6def7d46f9c89e2980309115b3c2c41" dependencies = [ "libc", + "linux-raw-sys", + "rustix", ] [[package]] @@ -6878,22 +6881,22 @@ checksum = "9828b178da53440fa9c766a3d2f73f7cf5d0ac1fe3980c1e5018d899fd19e07b" [[package]] name = "zerocopy" -version = "0.7.29" +version = "0.7.30" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "5d075cf85bbb114e933343e087b92f2146bac0d55b534cbb8188becf0039948e" +checksum = "306dca4455518f1f31635ec308b6b3e4eb1b11758cefafc782827d0aa7acb5c7" dependencies = [ "zerocopy-derive", ] [[package]] name = "zerocopy-derive" -version = "0.7.29" +version = "0.7.30" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "86cd5ca076997b97ef09d3ad65efe811fa68c9e874cb636ccb211223a813b0c2" +checksum = "be912bf68235a88fbefd1b73415cb218405958d1655b2ece9035a19920bdf6ba" dependencies = [ "proc-macro2", "quote", - "syn 2.0.39", + "syn 2.0.40", ] [[package]] @@ -6913,7 +6916,7 @@ checksum = "ce36e65b0d2999d2aafac989fb249189a141aee1f53c612c1f37d72631959f69" dependencies = [ "proc-macro2", "quote", - "syn 2.0.39", + "syn 2.0.40", ] [[package]] diff --git a/crates/jmap/src/api/http.rs b/crates/jmap/src/api/http.rs index 03ff7dac..17ee5bbe 100644 --- a/crates/jmap/src/api/http.rs +++ b/crates/jmap/src/api/http.rs @@ -365,7 +365,7 @@ pub async fn parse_jmap_request( }; } ("blob", "purge", &Method::GET) => { - return match jmap.store.blob_hash_purge(jmap.blob_store.clone()).await { + return match jmap.store.purge_blobs(jmap.blob_store.clone()).await { Ok(_) => { JsonResponse::new(Value::String("success".into())).into_http_response() } diff --git a/crates/jmap/src/lib.rs b/crates/jmap/src/lib.rs index d8762882..526ee7b4 100644 --- a/crates/jmap/src/lib.rs +++ b/crates/jmap/src/lib.rs @@ -863,17 +863,17 @@ pub enum NamedKey> { impl> From<&NamedKey> for ValueClass { fn from(key: &NamedKey) -> Self { match key { - NamedKey::Name(name) => ValueClass::Named { + NamedKey::Name(name) => ValueClass::Subspace { key: name.as_ref().to_vec(), id: 0, }, - NamedKey::Id(id) => ValueClass::Named { + NamedKey::Id(id) => ValueClass::Subspace { key: KeySerializer::new(std::mem::size_of::()) .write_leb128(*id) .finalize(), id: 1, }, - NamedKey::Quota(id) => ValueClass::Named { + NamedKey::Quota(id) => ValueClass::Subspace { key: KeySerializer::new(std::mem::size_of::()) .write_leb128(*id) .finalize(), @@ -883,7 +883,7 @@ impl> From<&NamedKey> for ValueClass { account_id, document_id, seq, - } => ValueClass::Named { + } => ValueClass::Subspace { key: KeySerializer::new(std::mem::size_of::() * 4) .write(*seq) .write(*account_id) diff --git a/crates/jmap/src/services/housekeeper.rs b/crates/jmap/src/services/housekeeper.rs index e173d233..b3e1af85 100644 --- a/crates/jmap/src/services/housekeeper.rs +++ b/crates/jmap/src/services/housekeeper.rs @@ -21,7 +21,7 @@ * for more details. */ -use std::{sync::Arc, time::Instant}; +use std::sync::Arc; use tokio::sync::mpsc; use utils::{ @@ -35,7 +35,6 @@ use crate::JMAP; use super::IPC_CHANNEL_BUFFER; pub enum Event { - PurgeDb, PurgeSessions, IndexStart, IndexDone, @@ -44,13 +43,7 @@ pub enum Event { Exit, } -const TASK_PURGE_DB: usize = 0; -const TASK_PURGE_SESSIONS: usize = 1; - pub fn spawn_housekeeper(core: Arc, settings: &Config, mut rx: mpsc::Receiver) { - let purge_db_at = settings - .property_or_static::("jmap.purge.schedule.db", "0 3 *") - .failed("Initialize housekeeper"); let purge_cache = settings .property_or_static::("jmap.purge.schedule.sessions", "15 * *") .failed("Initialize housekeeper"); @@ -68,15 +61,14 @@ pub fn spawn_housekeeper(core: Arc, settings: &Config, mut rx: mpsc::Recei }); loop { - let time_to_next = [purge_db_at.time_to_next(), purge_cache.time_to_next()]; - let mut tasks_to_run = [false, false]; - let start_time = Instant::now(); + let time_to_next = purge_cache.time_to_next(); + let mut do_purge = false; - match tokio::time::timeout(time_to_next.iter().min().copied().unwrap(), rx.recv()).await - { + match tokio::time::timeout(time_to_next, rx.recv()).await { Ok(Some(event)) => match event { - Event::PurgeDb => tasks_to_run[TASK_PURGE_DB] = true, - Event::PurgeSessions => tasks_to_run[TASK_PURGE_SESSIONS] = true, + Event::PurgeSessions => { + do_purge = true; + } Event::IndexStart => { if !index_busy { index_busy = true; @@ -115,49 +107,17 @@ pub fn spawn_housekeeper(core: Arc, settings: &Config, mut rx: mpsc::Recei Err(_) => (), } - // Check which tasks are due for execution - let now = Instant::now(); - for (pos, time_to_next) in time_to_next.into_iter().enumerate() { - if start_time + time_to_next <= now { - tasks_to_run[pos] = true; - } - } - - // Spawn tasks - for (task_id, do_run) in tasks_to_run.into_iter().enumerate() { - if !do_run { - continue; - } - + if do_purge { let core = core.clone(); - tokio::spawn(async move { - match task_id { - TASK_PURGE_DB => { - tracing::info!("Purging database..."); - if let Err(err) = core.store.purge_bitmaps().await { - tracing::error!("Error while purging bitmaps: {}", err); - } - - tracing::info!("Purging blobs...",); - if let Err(err) = - core.store.blob_hash_purge(core.blob_store.clone()).await - { - tracing::error!("Error while purging blobs: {}", err); - } - } - TASK_PURGE_SESSIONS => { - tracing::info!("Purging session cache."); - core.sessions.cleanup(); - core.access_tokens.cleanup(); - core.oauth_codes.cleanup(); - core.rate_limit_auth - .retain(|_, limiter| limiter.lock().is_active()); - core.rate_limit_unauth - .retain(|_, limiter| limiter.lock().is_active()); - } - _ => unreachable!(), - } + tracing::info!("Purging session cache."); + core.sessions.cleanup(); + core.access_tokens.cleanup(); + core.oauth_codes.cleanup(); + core.rate_limit_auth + .retain(|_, limiter| limiter.lock().is_active()); + core.rate_limit_unauth + .retain(|_, limiter| limiter.lock().is_active()); }); } } diff --git a/crates/main/src/main.rs b/crates/main/src/main.rs index ce1527b3..7f33c389 100644 --- a/crates/main/src/main.rs +++ b/crates/main/src/main.rs @@ -50,6 +50,14 @@ async fn main() -> std::io::Result<()> { let directory = config .parse_directory(&stores) .failed("Invalid configuration"); + let schedulers = config + .parse_purge_schedules( + &stores, + config.value("jmap.store.data"), + config.value("jmap.store.blob"), + ) + .await + .failed("Invalid configuration"); // Bind ports and drop privileges servers.bind(&config); @@ -77,7 +85,7 @@ async fn main() -> std::io::Result<()> { .failed("Invalid configuration file"); // Spawn servers - let (shutdown_tx, _shutdown_rx) = servers.spawn(|server, shutdown_rx| { + let (shutdown_tx, shutdown_rx) = servers.spawn(|server, shutdown_rx| { match &server.protocol { ServerProtocol::Smtp | ServerProtocol::Lmtp => { server.spawn(SmtpSessionManager::new(smtp.clone()), shutdown_rx) @@ -99,6 +107,11 @@ async fn main() -> std::io::Result<()> { }; }); + // Spawn purge schedulers + for scheduler in schedulers { + scheduler.spawn(shutdown_rx.clone()); + } + // Wait for shutdown signal wait_for_shutdown(&format!( "Shutting down Stalwart Mail Server v{}...", diff --git a/crates/nlp/src/bayes/mod.rs b/crates/nlp/src/bayes/mod.rs index 96aefa51..5a474ec7 100644 --- a/crates/nlp/src/bayes/mod.rs +++ b/crates/nlp/src/bayes/mod.rs @@ -105,3 +105,18 @@ impl std::hash::Hash for TokenHash { } impl nohash::IsEnabled for TokenHash {} + +impl From for Weights { + fn from(value: i64) -> Self { + Weights { + spam: value as u32, + ham: (value >> 32) as u32, + } + } +} + +impl From for i64 { + fn from(value: Weights) -> Self { + (value.ham as i64) << 32 | value.spam as i64 + } +} diff --git a/crates/smtp/Cargo.toml b/crates/smtp/Cargo.toml index 64d5697a..fb2d7f72 100644 --- a/crates/smtp/Cargo.toml +++ b/crates/smtp/Cargo.toml @@ -46,7 +46,7 @@ lru-cache = "0.1.2" rand = "0.8.5" x509-parser = "0.15.0" reqwest = { version = "0.11", default-features = false, features = ["rustls-tls-webpki-roots", "blocking"] } -serde = { version = "1.0", features = ["derive"] } +serde = { version = "1.0", features = ["derive", "rc"] } serde_json = "1.0" num_cpus = "1.15.0" lazy_static = "1.4" @@ -56,6 +56,7 @@ idna = "0.5" decancer = "1.6.1" unicode-security = "0.1.0" infer = "0.15.0" +bincode = "1.3.1" [features] test_mode = [] diff --git a/crates/smtp/src/config/scripts.rs b/crates/smtp/src/config/scripts.rs index f1bf0286..580c2e52 100644 --- a/crates/smtp/src/config/scripts.rs +++ b/crates/smtp/src/config/scripts.rs @@ -32,7 +32,7 @@ use parking_lot::RwLock; use sieve::{compiler::grammar::Capability, Compiler, Runtime}; use crate::{ - core::{SieveConfig, SieveCore}, + core::SieveCore, scripts::{functions::register_functions, plugins::RegisterSievePlugins}, }; use utils::{ @@ -193,23 +193,21 @@ impl ConfigSieve for Config { .iter() .map(|(k, v)| (k.to_string(), v.clone().into())) .collect(), - config: SieveConfig { - from_addr: self - .value("sieve.trusted.from-addr") - .map(|a| a.to_string()) - .unwrap_or(format!("MAILER-DAEMON@{hostname}")), - from_name: self - .value("sieve.trusted.from-name") - .unwrap_or("Mailer Daemon") - .to_string(), - return_path: self - .value("sieve.trusted.return-path") - .unwrap_or_default() - .to_string(), - sign, - directories: ctx.directory.directories.clone(), - lookup_stores: ctx.stores.lookup_stores.clone(), - }, + lookup_stores: ctx.stores.lookup_stores.clone(), + directories: ctx.directory.directories.clone(), + from_addr: self + .value("sieve.trusted.from-addr") + .map(|a| a.to_string()) + .unwrap_or(format!("MAILER-DAEMON@{hostname}")), + from_name: self + .value("sieve.trusted.from-name") + .unwrap_or("Mailer Daemon") + .to_string(), + return_path: self + .value("sieve.trusted.return-path") + .unwrap_or_default() + .to_string(), + sign, }) } } diff --git a/crates/smtp/src/core/mod.rs b/crates/smtp/src/core/mod.rs index ddf000e7..db50c672 100644 --- a/crates/smtp/src/core/mod.rs +++ b/crates/smtp/src/core/mod.rs @@ -112,17 +112,14 @@ pub struct SMTP { pub struct SieveCore { pub runtime: Runtime, pub scripts: AHashMap>, - pub lookup: AHashMap, - pub config: SieveConfig, -} -pub struct SieveConfig { pub from_addr: String, pub from_name: String, pub return_path: String, pub sign: Vec>, pub directories: AHashMap>, pub lookup_stores: AHashMap, + pub lookup: AHashMap, } pub struct Resolvers { diff --git a/crates/smtp/src/scripts/event_loop.rs b/crates/smtp/src/scripts/event_loop.rs index 3750373e..7e846436 100644 --- a/crates/smtp/src/scripts/event_loop.rs +++ b/crates/smtp/src/scripts/event_loop.rs @@ -57,8 +57,8 @@ impl SMTP { .filter(params.message.as_deref().map_or(b"", |m| &m[..])) .with_vars_env(params.variables) .with_envelope_list(params.envelope) - .with_user_address(&self.sieve.config.from_addr) - .with_user_full_name(&self.sieve.config.from_name); + .with_user_address(&self.sieve.from_addr) + .with_user_full_name(&self.sieve.from_name); let mut input = Input::script("__script", script); let mut messages: Vec> = Vec::new(); @@ -147,10 +147,10 @@ impl SMTP { message_id, } => { // Build message - let return_path_lcase = self.sieve.config.return_path.to_lowercase(); + let return_path_lcase = self.sieve.return_path.to_lowercase(); let return_path_domain = return_path_lcase.domain_part().to_string(); let mut message = Message::new_boxed( - self.sieve.config.return_path.clone(), + self.sieve.return_path.clone(), return_path_lcase, return_path_domain, ); @@ -278,9 +278,9 @@ impl SMTP { instance.message().raw_message().into() }; if let Some(raw_message) = raw_message { - let headers = if !self.sieve.config.sign.is_empty() { + let headers = if !self.sieve.sign.is_empty() { let mut headers = Vec::new(); - for dkim in &self.sieve.config.sign { + for dkim in &self.sieve.sign { match dkim.sign(raw_message) { Ok(signature) => { signature.write_header(&mut headers); diff --git a/crates/smtp/src/scripts/plugins/bayes.rs b/crates/smtp/src/scripts/plugins/bayes.rs index 668a82aa..b9154481 100644 --- a/crates/smtp/src/scripts/plugins/bayes.rs +++ b/crates/smtp/src/scripts/plugins/bayes.rs @@ -29,12 +29,12 @@ use nlp::{ tokenizers::osb::{OsbToken, OsbTokenizer}, }; use sieve::{runtime::Variable, FunctionMap}; -use store::Value; +use store::{write::key::KeySerializer, LookupKey, LookupStore, LookupValue, U64_LEN}; use tokio::runtime::Handle; -use crate::{config::scripts::SieveContext, core::Lookup}; +use crate::config::scripts::SieveContext; -use super::PluginContext; +use super::{lookup::VariableExists, PluginContext}; pub fn register_train(plugin_id: u32, fnc_map: &mut FunctionMap) { fnc_map.set_external_function("bayes_train", plugin_id, 3); @@ -61,17 +61,21 @@ pub fn exec_untrain(ctx: PluginContext<'_>) -> Variable { } fn train(ctx: PluginContext<'_>, is_train: bool) -> Variable { - let span = ctx.span; - let lookup_id = ctx.arguments[0].to_string(); - let lookup_train = if let Some(lookup_train) = ctx.core.sieve.lookup.get(lookup_id.as_ref()) { - lookup_train + let span: &tracing::Span = ctx.span; + let store = match &ctx.arguments[0] { + Variable::String(v) if !v.is_empty() => ctx.core.sieve.lookup_stores.get(v.as_ref()), + _ => ctx.core.sieve.lookup_stores.values().next(), + }; + + let store = if let Some(store) = store { + store } else { tracing::warn!( parent: span, context = "sieve:bayes_train", event = "failed", - reason = "Unknown lookup id", - lookup_id = %lookup_id, + reason = "Unknown store id", + lookup_store = ctx.arguments[0].to_string().as_ref(), ); return false.into(); }; @@ -102,44 +106,54 @@ fn train(ctx: PluginContext<'_>, is_train: bool) -> Variable { ); // Update weight and invalidate cache - for (hash, weights) in model.weights { - let (s_weight, h_weight) = if is_train { - (weights.spam as i64, weights.ham as i64) + if is_train { + for (hash, weights) in model.weights { + if handle + .block_on( + store.key_set( + KeySerializer::new(U64_LEN) + .write(hash.h1) + .write(hash.h2) + .finalize(), + LookupValue::Counter { + num: weights.into(), + }, + ), + ) + .is_err() + { + return false.into(); + } + ctx.bayes_cache.invalidate(&hash); + } + + // Update training counts + let weights = if is_spam { + Weights { spam: 1, ham: 0 } } else { - (-(weights.spam as i64), -(weights.ham as i64)) + Weights { spam: 0, ham: 1 } }; if handle - .block_on(lookup_train.lookup(vec![ - hash.h1.into(), - hash.h2.into(), - s_weight.into(), - h_weight.into(), - ])) - .is_none() + .block_on( + store.key_set( + KeySerializer::new(U64_LEN) + .write(0u64) + .write(0u64) + .finalize(), + LookupValue::Counter { + num: weights.into(), + }, + ), + ) + .is_err() { return false.into(); } - ctx.bayes_cache.invalidate(&hash); - } - - // Update training counts - let train_val = if is_train { 1i64 } else { -1i64 }; - let (spam_count, ham_count) = if is_spam { - (train_val, 0i64) } else { - (0i64, train_val) - }; - if handle - .block_on(lookup_train.query(vec![ - 0i64.into(), - 0i64.into(), - spam_count.into(), - ham_count.into(), - ])) - .is_none() - { + //TODO: Implement untrain return false.into(); } + ctx.bayes_cache.invalidate(&TokenHash::default()); true.into() @@ -147,20 +161,22 @@ fn train(ctx: PluginContext<'_>, is_train: bool) -> Variable { pub fn exec_classify(ctx: PluginContext<'_>) -> Variable { let span = ctx.span; - let lookup_id = ctx.arguments[0].to_string(); - let lookup_classify = - if let Some(lookup_classify) = ctx.core.sieve.lookup.get(lookup_id.as_ref()) { - lookup_classify - } else { - tracing::warn!( - parent: span, - context = "sieve:bayes_classify", - event = "failed", - reason = "Unknown lookup id", - lookup_id = %lookup_id, - ); - return Variable::default(); - }; + let store = match &ctx.arguments[0] { + Variable::String(v) if !v.is_empty() => ctx.core.sieve.lookup_stores.get(v.as_ref()), + _ => ctx.core.sieve.lookup_stores.values().next(), + }; + let store = if let Some(store) = store { + store + } else { + tracing::warn!( + parent: span, + context = "sieve:bayes_classify", + event = "failed", + reason = "Unknown store id", + lookup_id = ctx.arguments[0].to_string().as_ref(), + ); + return Variable::default(); + }; let text = ctx.arguments[1].to_string(); if text.is_empty() { return Variable::default(); @@ -189,7 +205,7 @@ pub fn exec_classify(ctx: PluginContext<'_>) -> Variable { // Obtain training counts let (spam_learns, ham_learns) = if let Some(weights) = ctx.bayes_cache - .get_or_update(TokenHash::default(), handle, lookup_classify) + .get_or_update(TokenHash::default(), handle, store) { (weights.spam, weights.ham) } else { @@ -220,9 +236,7 @@ pub fn exec_classify(ctx: PluginContext<'_>) -> Variable { OsbTokenizer::<_, TokenHash>::new(BayesTokenizer::new(text.as_ref(), &ctx.psl), 5) .filter_map(|t| { OsbToken { - inner: ctx - .bayes_cache - .get_or_update(t.inner, handle, lookup_classify)?, + inner: ctx.bayes_cache.get_or_update(t.inner, handle, store)?, idx: t.idx, } .into() @@ -246,20 +260,22 @@ pub fn exec_is_balanced(ctx: PluginContext<'_>) -> Variable { } let span = ctx.span; - let lookup_id = ctx.arguments[0].to_string(); - let lookup_classify = - if let Some(lookup_classify) = ctx.core.sieve.lookup.get(lookup_id.as_ref()) { - lookup_classify - } else { - tracing::warn!( - parent: span, - context = "sieve:bayes_is_balanced", - event = "failed", - reason = "Unknown lookup id", - lookup_id = %lookup_id, - ); - return Variable::default(); - }; + let store = match &ctx.arguments[0] { + Variable::String(v) if !v.is_empty() => ctx.core.sieve.lookup_stores.get(v.as_ref()), + _ => ctx.core.sieve.lookup_stores.values().next(), + }; + let store = if let Some(store) = store { + store + } else { + tracing::warn!( + parent: span, + context = "sieve:bayes_is_balanced", + event = "failed", + reason = "Unknown store id", + lookup_id = ctx.arguments[0].to_string().as_ref(), + ); + return Variable::default(); + }; let learn_spam = ctx.arguments[1].to_bool(); // Obtain training counts @@ -267,7 +283,7 @@ pub fn exec_is_balanced(ctx: PluginContext<'_>) -> Variable { let ctx = ctx.core.sieve.runtime.context(); let (spam_learns, ham_learns) = if let Some(weights) = ctx.bayes_cache - .get_or_update(TokenHash::default(), handle, lookup_classify) + .get_or_update(TokenHash::default(), handle, store) { (weights.spam as f64, weights.ham as f64) } else { @@ -308,7 +324,7 @@ trait LookupOrInsert { &self, hash: TokenHash, handle: &Handle, - get_token: &Lookup, + get_token: &LookupStore, ) -> Option; } @@ -317,27 +333,25 @@ impl LookupOrInsert for BayesTokenCache { &self, hash: TokenHash, handle: &Handle, - get_token: &Lookup, + get_token: &LookupStore, ) -> Option { if let Some(weights) = self.get(&hash) { weights.unwrap_or_default().into() - } else if let Some(result) = - handle.block_on(get_token.query(vec![hash.h1.into(), hash.h2.into()])) - { - let mut result = result.into_iter(); - match (result.next(), result.next()) { - (Some(Value::Integer(spam)), Some(Value::Integer(ham))) => { - let weights = Weights { - spam: spam as u32, - ham: ham as u32, - }; - self.insert_positive(hash, weights); - weights - } - _ => { - self.insert_negative(hash); - Weights::default() - } + } else if let Ok(result) = handle.block_on( + get_token.key_get::(LookupKey::Counter( + KeySerializer::new(U64_LEN) + .write(hash.h1) + .write(hash.h2) + .finalize(), + )), + ) { + if let LookupValue::Counter { num } = result { + let weights = Weights::from(num); + self.insert_positive(hash, weights); + weights + } else { + self.insert_negative(hash); + Weights::default() } .into() } else { diff --git a/crates/smtp/src/scripts/plugins/lookup.rs b/crates/smtp/src/scripts/plugins/lookup.rs index 01df16dd..615d4935 100644 --- a/crates/smtp/src/scripts/plugins/lookup.rs +++ b/crates/smtp/src/scripts/plugins/lookup.rs @@ -29,6 +29,7 @@ use std::{ use mail_auth::flate2; use sieve::{runtime::Variable, FunctionMap}; +use store::{Deserialize, LookupKey, LookupValue}; use crate::{ config::scripts::{RemoteList, SieveContext}, @@ -39,15 +40,19 @@ use crate::{ use super::PluginContext; pub fn register(plugin_id: u32, fnc_map: &mut FunctionMap) { - fnc_map.set_external_function("lookup", plugin_id, 2); + fnc_map.set_external_function("key_exists", plugin_id, 2); } -pub fn register_map(plugin_id: u32, fnc_map: &mut FunctionMap) { - fnc_map.set_external_function("lookup_map", plugin_id, 2); +pub fn register_get(plugin_id: u32, fnc_map: &mut FunctionMap) { + fnc_map.set_external_function("key_get", plugin_id, 2); +} + +pub fn register_set(plugin_id: u32, fnc_map: &mut FunctionMap) { + fnc_map.set_external_function("key_set", plugin_id, 4); } pub fn register_remote(plugin_id: u32, fnc_map: &mut FunctionMap) { - fnc_map.set_external_function("lookup_remote", plugin_id, 3); + fnc_map.set_external_function("key_exists_http", plugin_id, 3); } pub fn register_local_domain(plugin_id: u32, fnc_map: &mut FunctionMap) { @@ -55,16 +60,48 @@ pub fn register_local_domain(plugin_id: u32, fnc_map: &mut FunctionMap) -> Variable { - let lookup_id = ctx.arguments[0].to_string(); - let span = ctx.span; - if let Some(lookup) = ctx.core.sieve.lookup.get(lookup_id.as_ref()) { + let store = match &ctx.arguments[0] { + Variable::String(v) if v.contains('/') => { + if let Some(lookup) = ctx.core.sieve.lookup.get(v.as_ref()) { + return match &ctx.arguments[1] { + Variable::Array(items) => { + for item in items.iter() { + if !item.is_empty() + && ctx + .handle + .block_on(lookup.contains(to_store_value(item))) + .unwrap_or(false) + { + return true.into(); + } + } + false + } + v if !v.is_empty() => ctx + .handle + .block_on(lookup.contains(to_store_value(v))) + .unwrap_or(false), + _ => false, + } + .into(); + } + None + } + Variable::String(v) if !v.is_empty() => ctx.core.sieve.lookup_stores.get(v.as_ref()), + _ => ctx.core.sieve.lookup_stores.values().next(), + }; + + if let Some(store) = store { match &ctx.arguments[1] { Variable::Array(items) => { for item in items.iter() { if !item.is_empty() && ctx .handle - .block_on(lookup.contains(to_store_value(item))) + .block_on(store.key_get::(LookupKey::Key( + item.to_string().into_owned().into_bytes(), + ))) + .map(|v| v != LookupValue::None) .unwrap_or(false) { return true.into(); @@ -74,50 +111,109 @@ pub fn exec(ctx: PluginContext<'_>) -> Variable { } v if !v.is_empty() => ctx .handle - .block_on(lookup.contains(to_store_value(v))) + .block_on(store.key_get::(LookupKey::Key( + v.to_string().into_owned().into_bytes(), + ))) + .map(|v| v != LookupValue::None) .unwrap_or(false), _ => false, } } else { tracing::warn!( - parent: span, + parent: ctx.span, context = "sieve:lookup", event = "failed", reason = "Unknown lookup id", - lookup_id = %lookup_id, + lookup_id = ctx.arguments[0].to_string().as_ref(), ); false } .into() } -pub fn exec_map(ctx: PluginContext<'_>) -> Variable { - let lookup_id = ctx.arguments[0].to_string(); - let items = match &ctx.arguments[1] { - Variable::Array(l) => l.iter().map(to_store_value).collect(), - v if !v.is_empty() => vec![to_store_value(v)], - _ => vec![], - }; - let span = ctx.span; - - if !lookup_id.is_empty() && !items.is_empty() { - if let Some(lookup) = ctx.core.sieve.lookup.get(lookup_id.as_ref()) { - return ctx - .handle - .block_on(lookup.lookup(items)) - .unwrap_or_default(); - } else { - tracing::warn!( - parent: span, - context = "sieve:lookup", - event = "failed", - reason = "Unknown lookup id", - lookup_id = %lookup_id, - ); +pub fn exec_get(ctx: PluginContext<'_>) -> Variable { + let store = match &ctx.arguments[0] { + Variable::String(v) if v.contains('/') => { + if let Some(lookup) = ctx.core.sieve.lookup.get(v.as_ref()) { + let items = match &ctx.arguments[1] { + Variable::Array(l) => l.iter().map(to_store_value).collect(), + v if !v.is_empty() => vec![to_store_value(v)], + _ => vec![], + }; + return if !items.is_empty() { + ctx.handle + .block_on(lookup.lookup(items)) + .unwrap_or_default() + } else { + Variable::default() + }; + } + None } - } + Variable::String(v) if !v.is_empty() => ctx.core.sieve.lookup_stores.get(v.as_ref()), + _ => ctx.core.sieve.lookup_stores.values().next(), + }; - Variable::default() + if let Some(store) = store { + ctx.handle + .block_on(store.key_get::(LookupKey::Key( + ctx.arguments[1].to_string().into_owned().into_bytes(), + ))) + .map(|v| match v { + LookupValue::Value { value, .. } => value.into_inner(), + LookupValue::Counter { num } => num.into(), + LookupValue::None => Variable::default(), + }) + .unwrap_or_default() + } else { + tracing::warn!( + parent: ctx.span, + context = "sieve:key_get", + event = "failed", + reason = "Unknown store or lookup id", + lookup_id = ctx.arguments[0].to_string().as_ref(), + ); + Variable::default() + } +} + +pub fn exec_set(ctx: PluginContext<'_>) -> Variable { + let store = match &ctx.arguments[0] { + Variable::String(v) if !v.is_empty() => ctx.core.sieve.lookup_stores.get(v.as_ref()), + _ => ctx.core.sieve.lookup_stores.values().next(), + }; + + if let Some(store) = store { + let expires = match &ctx.arguments[3] { + Variable::Integer(v) => *v as u64, + Variable::Float(v) => *v as u64, + _ => 0, + }; + + ctx.handle + .block_on(store.key_set( + ctx.arguments[1].to_string().into_owned().into_bytes(), + LookupValue::Value { + value: if !ctx.arguments[2].is_empty() { + bincode::serialize(&ctx.arguments[2]).unwrap_or_default() + } else { + vec![] + }, + expires, + }, + )) + .is_ok() + .into() + } else { + tracing::warn!( + parent: ctx.span, + context = "sieve:key_set", + event = "failed", + reason = "Unknown store id", + store_id = ctx.arguments[0].to_string().as_ref(), + ); + Variable::default() + } } pub fn exec_remote(ctx: PluginContext<'_>) -> Variable { @@ -290,7 +386,7 @@ pub fn exec_remote(ctx: PluginContext<'_>) -> Variable { Err(err) => { tracing::warn!( parent: ctx.span, - context = "sieve:lookup_remote", + context = "sieve:key_exists_http", event = "failed", resource = resource.as_ref(), reason = %err, @@ -306,7 +402,7 @@ pub fn exec_remote(ctx: PluginContext<'_>) -> Variable { tracing::debug!( parent: ctx.span, - context = "sieve:lookup_remote", + context = "sieve:key_exists_http", event = "fetch", resource = resource.as_ref(), num_entries = list.entries.len(), @@ -319,7 +415,7 @@ pub fn exec_remote(ctx: PluginContext<'_>) -> Variable { Ok(Err(response)) => { tracing::warn!( parent: ctx.span, - context = "sieve:lookup_remote", + context = "sieve:key_exists_http", event = "failed", resource = resource.as_ref(), status = %response.status(), @@ -328,7 +424,7 @@ pub fn exec_remote(ctx: PluginContext<'_>) -> Variable { Err(err) => { tracing::warn!( parent: ctx.span, - context = "sieve:lookup_remote", + context = "sieve:key_exists_http", event = "failed", resource = resource.as_ref(), reason = %err, @@ -342,14 +438,18 @@ pub fn exec_remote(ctx: PluginContext<'_>) -> Variable { } pub fn exec_local_domain(ctx: PluginContext<'_>) -> Variable { - let directory_id = ctx.arguments[0].to_string(); let domain = ctx.arguments[0].to_string(); - if !directory_id.is_empty() && !domain.is_empty() { - if let Some(dir) = ctx.core.sieve.config.directories.get(directory_id.as_ref()) { + if !domain.is_empty() { + let directory = match &ctx.arguments[0] { + Variable::String(v) if !v.is_empty() => ctx.core.sieve.directories.get(v.as_ref()), + _ => ctx.core.sieve.directories.values().next(), + }; + + if let Some(directory) = directory { return ctx .handle - .block_on(dir.is_local_domain(domain.as_ref())) + .block_on(directory.is_local_domain(domain.as_ref())) .unwrap_or_default() .into(); } else { @@ -358,10 +458,38 @@ pub fn exec_local_domain(ctx: PluginContext<'_>) -> Variable { context = "sieve:is_local_domain", event = "failed", reason = "Unknown directory", - lookup_id = %directory_id, + lookup_id = ctx.arguments[0].to_string().as_ref(), ); } } Variable::default() } + +#[derive(Debug, PartialEq, Eq)] +pub(super) struct VariableWrapper(Variable); + +#[derive(Debug, PartialEq, Eq)] +pub(super) struct VariableExists; + +impl Deserialize for VariableWrapper { + fn deserialize(bytes: &[u8]) -> store::Result { + Ok(VariableWrapper( + bincode::deserialize::(bytes).unwrap_or_else(|_| { + Variable::String(String::from_utf8_lossy(bytes).into_owned().into()) + }), + )) + } +} + +impl Deserialize for VariableExists { + fn deserialize(_: &[u8]) -> store::Result { + Ok(VariableExists) + } +} + +impl VariableWrapper { + pub fn into_inner(self) -> Variable { + self.0 + } +} diff --git a/crates/smtp/src/scripts/plugins/mod.rs b/crates/smtp/src/scripts/plugins/mod.rs index 5099913e..dfb2babb 100644 --- a/crates/smtp/src/scripts/plugins/mod.rs +++ b/crates/smtp/src/scripts/plugins/mod.rs @@ -50,11 +50,12 @@ pub struct PluginContext<'x> { pub arguments: Vec, } -const PLUGINS_EXEC: [ExecPluginFnc; 15] = [ +const PLUGINS_EXEC: [ExecPluginFnc; 16] = [ query::exec, exec::exec, lookup::exec, - lookup::exec_map, + lookup::exec_get, + lookup::exec_set, lookup::exec_remote, lookup::exec_local_domain, dns::exec, @@ -67,11 +68,12 @@ const PLUGINS_EXEC: [ExecPluginFnc; 15] = [ pyzor::exec, headers::exec, ]; -const PLUGINS_REGISTER: [RegisterPluginFnc; 15] = [ +const PLUGINS_REGISTER: [RegisterPluginFnc; 16] = [ query::register, exec::register, lookup::register, - lookup::register_map, + lookup::register_get, + lookup::register_set, lookup::register_remote, lookup::register_local_domain, dns::register, diff --git a/crates/smtp/src/scripts/plugins/query.rs b/crates/smtp/src/scripts/plugins/query.rs index bdb52ebf..98905b12 100644 --- a/crates/smtp/src/scripts/plugins/query.rs +++ b/crates/smtp/src/scripts/plugins/query.rs @@ -41,7 +41,7 @@ pub fn exec(ctx: PluginContext<'_>) -> Variable { // Obtain store name let store = ctx.arguments[0].to_string(); - let store = if let Some(store_) = ctx.core.sieve.config.lookup_stores.get(store.as_ref()) { + let store = if let Some(store_) = ctx.core.sieve.lookup_stores.get(store.as_ref()) { store_ } else { tracing::warn!( diff --git a/crates/store/src/config.rs b/crates/store/src/config.rs index 8763a602..f67de2a4 100644 --- a/crates/store/src/config.rs +++ b/crates/store/src/config.rs @@ -24,10 +24,11 @@ use std::sync::Arc; use async_trait::async_trait; -use utils::config::{utils::AsKey, Config}; +use utils::config::{cron::SimpleCron, utils::AsKey, Config}; use crate::{ backend::{fs::FsStore, memory::MemoryStore}, + write::purge::{PurgeSchedule, PurgeStore}, Lookup, LookupStore, Store, Stores, }; @@ -55,6 +56,12 @@ use crate::backend::elastic::ElasticSearchStore; #[async_trait] pub trait ConfigStore { async fn parse_stores(&self) -> utils::config::Result; + async fn parse_purge_schedules( + &self, + stores: &Stores, + store: Option<&str>, + blob_store: Option<&str>, + ) -> utils::config::Result>; } #[async_trait] @@ -78,7 +85,10 @@ impl ConfigStore for Config { config .fts_stores .insert(store_id.clone(), db.clone().into()); - config.blob_stores.insert(store_id.clone(), db.into()); + config + .blob_stores + .insert(store_id.clone(), db.clone().into()); + config.lookup_stores.insert(store_id, db.into()); continue; } #[cfg(feature = "foundation")] @@ -88,7 +98,10 @@ impl ConfigStore for Config { config .fts_stores .insert(store_id.clone(), db.clone().into()); - config.blob_stores.insert(store_id.clone(), db.into()); + config + .blob_stores + .insert(store_id.clone(), db.clone().into()); + config.lookup_stores.insert(store_id, db.into()); continue; } #[cfg(feature = "postgres")] @@ -196,6 +209,60 @@ impl ConfigStore for Config { Ok(config) } + + async fn parse_purge_schedules( + &self, + stores: &Stores, + store_id: Option<&str>, + blob_store_id: Option<&str>, + ) -> utils::config::Result> { + let mut schedules = Vec::new(); + + if let Some(store) = store_id.and_then(|store_id| stores.stores.get(store_id)) { + let store_id = store_id.unwrap(); + if let Some(cron) = + self.property::(("store", store_id, "purge.frequency"))? + { + schedules.push(PurgeSchedule { + cron, + store_id: store_id.to_string(), + store: PurgeStore::Bitmaps(store.clone()), + }); + } + + if let Some(blob_store) = + blob_store_id.and_then(|blob_store_id| stores.blob_stores.get(blob_store_id)) + { + let blob_store_id = blob_store_id.unwrap(); + if let Some(cron) = + self.property::(("store", blob_store_id, "purge.frequency"))? + { + schedules.push(PurgeSchedule { + cron, + store_id: blob_store_id.to_string(), + store: PurgeStore::Blobs { + store: store.clone(), + blob_store: blob_store.clone(), + }, + }); + } + } + } + + for (store_id, store) in &stores.lookup_stores { + if let Some(cron) = + self.property::(("store", store_id.as_str(), "purge.frequency"))? + { + schedules.push(PurgeSchedule { + cron, + store_id: store_id.clone(), + store: PurgeStore::Lookup(store.clone()), + }); + } + } + + Ok(schedules) + } } impl From for String { diff --git a/crates/store/src/dispatch/blob.rs b/crates/store/src/dispatch/blob.rs new file mode 100644 index 00000000..005ac17d --- /dev/null +++ b/crates/store/src/dispatch/blob.rs @@ -0,0 +1,88 @@ +/* + * Copyright (c) 2023 Stalwart Labs Ltd. + * + * This file is part of the Stalwart Mail Server. + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU Affero General Public License as + * published by the Free Software Foundation, either version 3 of + * the License, or (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU Affero General Public License for more details. + * in the LICENSE file at the top-level directory of this distribution. + * You should have received a copy of the GNU Affero General Public License + * along with this program. If not, see . + * + * You can be released from the requirements of the AGPLv3 license by + * purchasing a commercial license. Please contact licensing@stalw.art + * for more details. +*/ + +use std::ops::Range; + +use crate::{BlobStore, Store}; + +impl BlobStore { + pub async fn get_blob(&self, key: &[u8], range: Range) -> crate::Result>> { + match self { + Self::Store(store) => match store { + #[cfg(feature = "sqlite")] + Store::SQLite(store) => store.get_blob(key, range).await, + #[cfg(feature = "foundation")] + Store::FoundationDb(store) => store.get_blob(key, range).await, + #[cfg(feature = "postgres")] + Store::PostgreSQL(store) => store.get_blob(key, range).await, + #[cfg(feature = "mysql")] + Store::MySQL(store) => store.get_blob(key, range).await, + #[cfg(feature = "rocks")] + Store::RocksDb(store) => store.get_blob(key, range).await, + }, + Self::Fs(store) => store.get_blob(key, range).await, + #[cfg(feature = "s3")] + Self::S3(store) => store.get_blob(key, range).await, + } + } + + pub async fn put_blob(&self, key: &[u8], data: &[u8]) -> crate::Result<()> { + match self { + Self::Store(store) => match store { + #[cfg(feature = "sqlite")] + Store::SQLite(store) => store.put_blob(key, data).await, + #[cfg(feature = "foundation")] + Store::FoundationDb(store) => store.put_blob(key, data).await, + #[cfg(feature = "postgres")] + Store::PostgreSQL(store) => store.put_blob(key, data).await, + #[cfg(feature = "mysql")] + Store::MySQL(store) => store.put_blob(key, data).await, + #[cfg(feature = "rocks")] + Store::RocksDb(store) => store.put_blob(key, data).await, + }, + Self::Fs(store) => store.put_blob(key, data).await, + #[cfg(feature = "s3")] + Self::S3(store) => store.put_blob(key, data).await, + } + } + + pub async fn delete_blob(&self, key: &[u8]) -> crate::Result { + match self { + Self::Store(store) => match store { + #[cfg(feature = "sqlite")] + Store::SQLite(store) => store.delete_blob(key).await, + #[cfg(feature = "foundation")] + Store::FoundationDb(store) => store.delete_blob(key).await, + #[cfg(feature = "postgres")] + Store::PostgreSQL(store) => store.delete_blob(key).await, + #[cfg(feature = "mysql")] + Store::MySQL(store) => store.delete_blob(key).await, + #[cfg(feature = "rocks")] + Store::RocksDb(store) => store.delete_blob(key).await, + }, + Self::Fs(store) => store.delete_blob(key).await, + #[cfg(feature = "s3")] + Self::S3(store) => store.delete_blob(key).await, + } + } +} diff --git a/crates/store/src/dispatch/fts.rs b/crates/store/src/dispatch/fts.rs new file mode 100644 index 00000000..3337c515 --- /dev/null +++ b/crates/store/src/dispatch/fts.rs @@ -0,0 +1,82 @@ +/* + * Copyright (c) 2023 Stalwart Labs Ltd. + * + * This file is part of the Stalwart Mail Server. + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU Affero General Public License as + * published by the Free Software Foundation, either version 3 of + * the License, or (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU Affero General Public License for more details. + * in the LICENSE file at the top-level directory of this distribution. + * You should have received a copy of the GNU Affero General Public License + * along with this program. If not, see . + * + * You can be released from the requirements of the AGPLv3 license by + * purchasing a commercial license. Please contact licensing@stalw.art + * for more details. +*/ + +use std::fmt::Display; + +use roaring::RoaringBitmap; + +use crate::{ + fts::{index::FtsDocument, FtsFilter}, + FtsStore, +}; + +impl FtsStore { + pub async fn index + Display + Clone + std::fmt::Debug>( + &self, + document: FtsDocument<'_, T>, + ) -> crate::Result<()> { + match self { + FtsStore::Store(store) => store.fts_index(document).await, + #[cfg(feature = "elastic")] + FtsStore::ElasticSearch(store) => store.fts_index(document).await, + } + } + + pub async fn query + Display + Clone + std::fmt::Debug>( + &self, + account_id: u32, + collection: impl Into, + filters: Vec>, + ) -> crate::Result { + match self { + FtsStore::Store(store) => store.fts_query(account_id, collection, filters).await, + #[cfg(feature = "elastic")] + FtsStore::ElasticSearch(store) => { + store.fts_query(account_id, collection, filters).await + } + } + } + + pub async fn remove( + &self, + account_id: u32, + collection: u8, + document_id: u32, + ) -> crate::Result { + match self { + FtsStore::Store(store) => store.fts_remove(account_id, collection, document_id).await, + #[cfg(feature = "elastic")] + FtsStore::ElasticSearch(store) => { + store.fts_remove(account_id, collection, document_id).await + } + } + } + + pub async fn remove_all(&self, account_id: u32) -> crate::Result<()> { + match self { + FtsStore::Store(store) => store.fts_remove_all(account_id).await, + #[cfg(feature = "elastic")] + FtsStore::ElasticSearch(store) => store.fts_remove_all(account_id).await, + } + } +} diff --git a/crates/store/src/dispatch/lookup.rs b/crates/store/src/dispatch/lookup.rs new file mode 100644 index 00000000..2b848b0c --- /dev/null +++ b/crates/store/src/dispatch/lookup.rs @@ -0,0 +1,182 @@ +/* + * Copyright (c) 2023 Stalwart Labs Ltd. + * + * This file is part of the Stalwart Mail Server. + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU Affero General Public License as + * published by the Free Software Foundation, either version 3 of + * the License, or (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU Affero General Public License for more details. + * in the LICENSE file at the top-level directory of this distribution. + * You should have received a copy of the GNU Affero General Public License + * along with this program. If not, see . + * + * You can be released from the requirements of the AGPLv3 license by + * purchasing a commercial license. Please contact licensing@stalw.art + * for more details. +*/ + +use crate::{ + write::{ + key::{DeserializeBigEndian, KeySerializer}, + now, BatchBuilder, Operation, ValueClass, ValueOp, + }, + Deserialize, IterateParams, LookupKey, LookupStore, LookupValue, QueryResult, Store, Value, + ValueKey, U64_LEN, +}; + +impl LookupStore { + pub async fn query( + &self, + query: &str, + params: Vec>, + ) -> crate::Result { + let result = match self { + LookupStore::Store(store) => match store { + #[cfg(feature = "sqlite")] + Store::SQLite(store) => store.query(query, params).await, + #[cfg(feature = "postgres")] + Store::PostgreSQL(store) => store.query(query, params).await, + #[cfg(feature = "mysql")] + Store::MySQL(store) => store.query(query, params).await, + _ => Err(crate::Error::InternalError( + "Store does not support queries".into(), + )), + }, + LookupStore::Memory(store) => store.query(query, params), + }; + + tracing::trace!( context = "store", event = "query", query = query, result = ?result); + + result + } + + pub async fn key_set(&self, key: Vec, value: LookupValue>) -> crate::Result<()> { + match self { + LookupStore::Store(store) => { + let (class, op) = match value { + LookupValue::Value { value, expires } => ( + ValueClass::Key { key }, + ValueOp::Set( + KeySerializer::new(value.len() + U64_LEN) + .write(if expires > 0 { + now() + expires + } else { + u64::MAX + }) + .write(value.as_slice()) + .finalize(), + ), + ), + LookupValue::Counter { num } => (ValueClass::Key { key }, ValueOp::Add(num)), + LookupValue::None => return Ok(()), + }; + + let mut batch = BatchBuilder::new(); + batch.ops.push(Operation::Value { class, op }); + store.write(batch.build()).await + } + LookupStore::Memory(_) => unimplemented!(), + } + } + + pub async fn key_get( + &self, + key: LookupKey, + ) -> crate::Result> { + match self { + LookupStore::Store(store) => match key { + LookupKey::Key(key) => store + .get_value::>(ValueKey { + account_id: 0, + collection: 0, + document_id: 0, + class: ValueClass::Key { key }, + }) + .await + .map(|value| value.unwrap_or(LookupValue::None)), + LookupKey::Counter(key) => store + .get_counter(ValueKey { + account_id: 0, + collection: 0, + document_id: 0, + class: ValueClass::Key { key }, + }) + .await + .map(|num| LookupValue::Counter { num }), + }, + LookupStore::Memory(_) => unimplemented!(), + } + } + + pub async fn purge_expired(&self) -> crate::Result<()> { + match self { + LookupStore::Store(store) => { + let from_key = ValueKey { + account_id: 0, + collection: 0, + document_id: 0, + class: ValueClass::Key { key: vec![0u8] }, + }; + let to_key = ValueKey { + account_id: 0, + collection: 0, + document_id: 0, + class: ValueClass::Key { + key: vec![u8::MAX; 10], + }, + }; + + let current_time = now(); + let mut expired_keys = Vec::new(); + store + .iterate(IterateParams::new(from_key, to_key), |key, value| { + if value.deserialize_be_u64(0)? < current_time { + expired_keys.push(key.to_vec()); + } + Ok(true) + }) + .await?; + if !expired_keys.is_empty() { + let mut batch = BatchBuilder::new(); + for key in expired_keys { + batch.ops.push(Operation::Value { + class: ValueClass::Key { key }, + op: ValueOp::Clear, + }); + if batch.ops.len() >= 1000 { + store.write(batch.build()).await?; + batch = BatchBuilder::new(); + } + } + if !batch.ops.is_empty() { + store.write(batch.build()).await?; + } + } + } + LookupStore::Memory(_) => {} + } + + Ok(()) + } +} + +impl Deserialize for LookupValue { + fn deserialize(bytes: &[u8]) -> crate::Result { + bytes.deserialize_be_u64(0).and_then(|expires| { + Ok(if expires > now() { + LookupValue::Value { + value: T::deserialize(bytes.get(U64_LEN..).unwrap_or_default())?, + expires, + } + } else { + LookupValue::None + }) + }) + } +} diff --git a/crates/store/src/dispatch/mod.rs b/crates/store/src/dispatch/mod.rs new file mode 100644 index 00000000..dbc34c81 --- /dev/null +++ b/crates/store/src/dispatch/mod.rs @@ -0,0 +1,27 @@ +/* + * Copyright (c) 2023 Stalwart Labs Ltd. + * + * This file is part of the Stalwart Mail Server. + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU Affero General Public License as + * published by the Free Software Foundation, either version 3 of + * the License, or (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU Affero General Public License for more details. + * in the LICENSE file at the top-level directory of this distribution. + * You should have received a copy of the GNU Affero General Public License + * along with this program. If not, see . + * + * You can be released from the requirements of the AGPLv3 license by + * purchasing a commercial license. Please contact licensing@stalw.art + * for more details. +*/ + +pub mod blob; +pub mod fts; +pub mod lookup; +pub mod store; diff --git a/crates/store/src/dispatch.rs b/crates/store/src/dispatch/store.rs similarity index 75% rename from crates/store/src/dispatch.rs rename to crates/store/src/dispatch/store.rs index 67de2cde..3b2b770e 100644 --- a/crates/store/src/dispatch.rs +++ b/crates/store/src/dispatch/store.rs @@ -21,19 +21,14 @@ * for more details. */ -use std::{ - fmt::Display, - ops::{BitAndAssign, Range}, -}; +use std::ops::{BitAndAssign, Range}; use roaring::RoaringBitmap; use crate::{ - fts::{index::FtsDocument, FtsFilter}, write::{key::KeySerializer, Batch, BitmapClass, ValueClass}, - BitmapKey, BlobStore, Deserialize, FtsStore, IterateParams, Key, LookupStore, QueryResult, - Store, Value, ValueKey, SUBSPACE_BITMAPS, SUBSPACE_INDEXES, SUBSPACE_INDEX_VALUES, - SUBSPACE_LOGS, SUBSPACE_VALUES, U32_LEN, + BitmapKey, Deserialize, IterateParams, Key, Store, ValueKey, SUBSPACE_BITMAPS, + SUBSPACE_INDEXES, SUBSPACE_INDEX_VALUES, SUBSPACE_LOGS, SUBSPACE_VALUES, U32_LEN, }; impl Store { @@ -384,7 +379,7 @@ impl Store { use crate::{SUBSPACE_BLOBS, SUBSPACE_BLOB_DATA, SUBSPACE_COUNTERS}; self.blob_hash_expire_all().await; - self.blob_hash_purge(blob_store).await.unwrap(); + self.purge_blobs(blob_store).await.unwrap(); self.purge_bitmaps().await.unwrap(); let store = self.clone(); @@ -493,143 +488,3 @@ impl Store { } } } - -impl BlobStore { - pub async fn get_blob(&self, key: &[u8], range: Range) -> crate::Result>> { - match self { - Self::Store(store) => match store { - #[cfg(feature = "sqlite")] - Store::SQLite(store) => store.get_blob(key, range).await, - #[cfg(feature = "foundation")] - Store::FoundationDb(store) => store.get_blob(key, range).await, - #[cfg(feature = "postgres")] - Store::PostgreSQL(store) => store.get_blob(key, range).await, - #[cfg(feature = "mysql")] - Store::MySQL(store) => store.get_blob(key, range).await, - #[cfg(feature = "rocks")] - Store::RocksDb(store) => store.get_blob(key, range).await, - }, - Self::Fs(store) => store.get_blob(key, range).await, - #[cfg(feature = "s3")] - Self::S3(store) => store.get_blob(key, range).await, - } - } - - pub async fn put_blob(&self, key: &[u8], data: &[u8]) -> crate::Result<()> { - match self { - Self::Store(store) => match store { - #[cfg(feature = "sqlite")] - Store::SQLite(store) => store.put_blob(key, data).await, - #[cfg(feature = "foundation")] - Store::FoundationDb(store) => store.put_blob(key, data).await, - #[cfg(feature = "postgres")] - Store::PostgreSQL(store) => store.put_blob(key, data).await, - #[cfg(feature = "mysql")] - Store::MySQL(store) => store.put_blob(key, data).await, - #[cfg(feature = "rocks")] - Store::RocksDb(store) => store.put_blob(key, data).await, - }, - Self::Fs(store) => store.put_blob(key, data).await, - #[cfg(feature = "s3")] - Self::S3(store) => store.put_blob(key, data).await, - } - } - - pub async fn delete_blob(&self, key: &[u8]) -> crate::Result { - match self { - Self::Store(store) => match store { - #[cfg(feature = "sqlite")] - Store::SQLite(store) => store.delete_blob(key).await, - #[cfg(feature = "foundation")] - Store::FoundationDb(store) => store.delete_blob(key).await, - #[cfg(feature = "postgres")] - Store::PostgreSQL(store) => store.delete_blob(key).await, - #[cfg(feature = "mysql")] - Store::MySQL(store) => store.delete_blob(key).await, - #[cfg(feature = "rocks")] - Store::RocksDb(store) => store.delete_blob(key).await, - }, - Self::Fs(store) => store.delete_blob(key).await, - #[cfg(feature = "s3")] - Self::S3(store) => store.delete_blob(key).await, - } - } -} - -impl FtsStore { - pub async fn index + Display + Clone + std::fmt::Debug>( - &self, - document: FtsDocument<'_, T>, - ) -> crate::Result<()> { - match self { - FtsStore::Store(store) => store.fts_index(document).await, - #[cfg(feature = "elastic")] - FtsStore::ElasticSearch(store) => store.fts_index(document).await, - } - } - - pub async fn query + Display + Clone + std::fmt::Debug>( - &self, - account_id: u32, - collection: impl Into, - filters: Vec>, - ) -> crate::Result { - match self { - FtsStore::Store(store) => store.fts_query(account_id, collection, filters).await, - #[cfg(feature = "elastic")] - FtsStore::ElasticSearch(store) => { - store.fts_query(account_id, collection, filters).await - } - } - } - - pub async fn remove( - &self, - account_id: u32, - collection: u8, - document_id: u32, - ) -> crate::Result { - match self { - FtsStore::Store(store) => store.fts_remove(account_id, collection, document_id).await, - #[cfg(feature = "elastic")] - FtsStore::ElasticSearch(store) => { - store.fts_remove(account_id, collection, document_id).await - } - } - } - - pub async fn remove_all(&self, account_id: u32) -> crate::Result<()> { - match self { - FtsStore::Store(store) => store.fts_remove_all(account_id).await, - #[cfg(feature = "elastic")] - FtsStore::ElasticSearch(store) => store.fts_remove_all(account_id).await, - } - } -} - -impl LookupStore { - pub async fn query( - &self, - query: &str, - params: Vec>, - ) -> crate::Result { - let result = match self { - LookupStore::Store(store) => match store { - #[cfg(feature = "sqlite")] - Store::SQLite(store) => store.query(query, params).await, - #[cfg(feature = "postgres")] - Store::PostgreSQL(store) => store.query(query, params).await, - #[cfg(feature = "mysql")] - Store::MySQL(store) => store.query(query, params).await, - _ => Err(crate::Error::InternalError( - "Store does not support lookups".into(), - )), - }, - LookupStore::Memory(store) => store.query(query, params), - }; - - tracing::trace!( context = "store", event = "query", query = query, result = ?result); - - result - } -} diff --git a/crates/store/src/fts/index.rs b/crates/store/src/fts/index.rs index 6894b2b9..d00bbf15 100644 --- a/crates/store/src/fts/index.rs +++ b/crates/store/src/fts/index.rs @@ -249,8 +249,8 @@ impl Store { .with_collection(document.collection) .update_document(document.document_id); - for (pos, key) in keys.into_iter().enumerate() { - if pos > 0 && pos & 1023 == 0 { + for key in keys.into_iter() { + if batch.ops.len() >= 1000 { self.write(batch.build()).await?; batch = BatchBuilder::new(); batch @@ -296,8 +296,8 @@ impl Store { .with_collection(collection) .update_document(document_id); - for (pos, key) in term_index.ops.into_iter().enumerate() { - if pos > 0 && pos & 1023 == 0 { + for key in term_index.ops.into_iter() { + if batch.ops.len() >= 1000 { self.write(batch.build()).await?; batch = BatchBuilder::new(); batch diff --git a/crates/store/src/lib.rs b/crates/store/src/lib.rs index df313f82..56c0b831 100644 --- a/crates/store/src/lib.rs +++ b/crates/store/src/lib.rs @@ -318,6 +318,19 @@ impl From for LookupStore { } #[derive(Clone, Debug)] +pub enum LookupKey { + Key(Vec), + Counter(Vec), +} + +#[derive(Clone, Debug, PartialEq, Eq)] +pub enum LookupValue { + Value { value: T, expires: u64 }, + Counter { num: i64 }, + None, +} + +#[derive(Clone, Debug, PartialEq)] pub enum Value<'x> { Integer(i64), Bool(bool), @@ -327,6 +340,8 @@ pub enum Value<'x> { Null, } +impl Eq for Value<'_> {} + impl<'x> Value<'x> { pub fn to_str<'y: 'x>(&'y self) -> Cow<'x, str> { match self { diff --git a/crates/store/src/query/acl.rs b/crates/store/src/query/acl.rs index 68eaa79b..b7734341 100644 --- a/crates/store/src/query/acl.rs +++ b/crates/store/src/query/acl.rs @@ -128,8 +128,8 @@ impl Store { let mut batch = BatchBuilder::new(); batch.with_account_id(account_id); let mut last_collection = u8::MAX; - for (pos, (class, acl_item)) in delete_keys.into_iter().enumerate() { - if pos > 0 && pos & 511 == 0 { + for (class, acl_item) in delete_keys.into_iter() { + if batch.ops.len() >= 1000 { self.write(batch.build()).await?; batch = BatchBuilder::new(); batch.with_account_id(account_id); diff --git a/crates/store/src/write/blob.rs b/crates/store/src/write/blob.rs index d69b7ae8..b5c67c80 100644 --- a/crates/store/src/write/blob.rs +++ b/crates/store/src/write/blob.rs @@ -122,7 +122,10 @@ impl Store { account_id: *account_id, collection: 0, document_id: 0, - op: BlobOp::Reserve { until: 0, size: 0 }, + op: BlobOp::Reserve { + until: now(), + size: 0, + }, hash: hash.as_ref().clone(), }, BlobKey { @@ -131,7 +134,7 @@ impl Store { document_id: 0, op: BlobOp::Reserve { until: u64::MAX, - size: 0, + size: u32::MAX as usize, }, hash: hash.as_ref().clone(), }, @@ -175,7 +178,7 @@ impl Store { Ok(has_access) } - pub async fn blob_hash_purge(&self, blob_store: BlobStore) -> crate::Result<()> { + pub async fn purge_blobs(&self, blob_store: BlobStore) -> crate::Result<()> { // Remove expired temporary blobs let from_key = BlobKey { account_id: 0, @@ -284,8 +287,8 @@ impl Store { // Delete hashes let mut batch = BatchBuilder::new(); let mut last_account_id = u32::MAX; - for (pos, key) in delete_keys.into_iter().enumerate() { - if pos > 0 && pos & 511 == 0 { + for key in delete_keys.into_iter() { + if batch.ops.len() >= 1000 { last_account_id = u32::MAX; self.write(batch.build()).await?; batch = BatchBuilder::new(); @@ -353,8 +356,8 @@ impl Store { let mut batch = BatchBuilder::new(); batch.with_account_id(account_id); let mut last_collection = u8::MAX; - for (pos, key) in delete_keys.into_iter().enumerate() { - if pos > 0 && pos & 511 == 0 { + for key in delete_keys.into_iter() { + if batch.ops.len() >= 1000 { self.write(batch.build()).await?; batch = BatchBuilder::new(); batch.with_account_id(account_id); diff --git a/crates/store/src/write/key.rs b/crates/store/src/write/key.rs index 6a779c41..83f9c2d2 100644 --- a/crates/store/src/write/key.rs +++ b/crates/store/src/write/key.rs @@ -266,15 +266,14 @@ impl + Sync + Send> Key for ValueKey { .write(self.account_id) .write(self.collection) .write(self.document_id), - ValueClass::Ttl { key, expires } => if include_subspace { + ValueClass::Key { key } => if include_subspace { KeySerializer::new(key.len() + U64_LEN + 2).write(crate::SUBSPACE_INDEX_VALUES) } else { KeySerializer::new(key.len() + U64_LEN + 1) } .write(2u8) - .write(key.as_slice()) - .write(*expires), - ValueClass::Named { key, id } => if include_subspace { + .write(key.as_slice()), + ValueClass::Subspace { key, id } => if include_subspace { KeySerializer::new(key.len() + 2).write(crate::SUBSPACE_INDEX_VALUES) } else { KeySerializer::new(key.len() + 1) diff --git a/crates/store/src/write/mod.rs b/crates/store/src/write/mod.rs index ceeb0a5b..447faac2 100644 --- a/crates/store/src/write/mod.rs +++ b/crates/store/src/write/mod.rs @@ -45,6 +45,7 @@ pub mod blob; pub mod hash; pub mod key; pub mod log; +pub mod purge; #[cfg(not(feature = "test_mode"))] pub(crate) const ID_ASSIGNMENT_EXPIRY: u64 = 60 * 60; // seconds @@ -137,8 +138,8 @@ pub enum TagValue { pub enum ValueClass { Property(u8), Acl(u32), - Named { key: Vec, id: u8 }, - Ttl { key: Vec, expires: u64 }, + Subspace { key: Vec, id: u8 }, + Key { key: Vec }, TermIndex, ReservedId, } diff --git a/crates/store/src/write/purge.rs b/crates/store/src/write/purge.rs new file mode 100644 index 00000000..ecd8c1ed --- /dev/null +++ b/crates/store/src/write/purge.rs @@ -0,0 +1,77 @@ +/* + * Copyright (c) 2023 Stalwart Labs Ltd. + * + * This file is part of Stalwart Mail Server. + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU Affero General Public License as + * published by the Free Software Foundation, either version 3 of + * the License, or (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU Affero General Public License for more details. + * in the LICENSE file at the top-level directory of this distribution. + * You should have received a copy of the GNU Affero General Public License + * along with this program. If not, see . + * + * You can be released from the requirements of the AGPLv3 license by + * purchasing a commercial license. Please contact licensing@stalw.art + * for more details. +*/ + +use tokio::sync::watch; +use utils::config::cron::SimpleCron; + +use crate::{BlobStore, LookupStore, Store}; + +pub enum PurgeStore { + Bitmaps(Store), + Blobs { store: Store, blob_store: BlobStore }, + Lookup(LookupStore), +} + +pub struct PurgeSchedule { + pub cron: SimpleCron, + pub store_id: String, + pub store: PurgeStore, +} + +impl PurgeSchedule { + pub fn spawn(self, mut shutdown_rx: watch::Receiver) { + tracing::debug!("Purge task started for store {:?}.", self.store_id); + tokio::spawn(async move { + loop { + if tokio::time::timeout(self.cron.time_to_next(), shutdown_rx.changed()) + .await + .is_ok() + { + tracing::debug!("Purge task exiting for store {:?}.", self.store_id); + return; + } + + let result = match &self.store { + PurgeStore::Bitmaps(store) => store.purge_bitmaps().await, + PurgeStore::Blobs { store, blob_store } => { + store.purge_blobs(blob_store.clone()).await + } + PurgeStore::Lookup(store) => store.purge_expired().await, + }; + + if let Err(err) = result { + tracing::warn!( + "Purge {} task failed for store {:?}: {:?}", + match &self.store { + PurgeStore::Bitmaps(_) => "bitmaps", + PurgeStore::Blobs { .. } => "blobs", + PurgeStore::Lookup(_) => "expired keys", + }, + self.store_id, + err + ); + } + } + }); + } +} diff --git a/resources/config/spamfilter/scripts/bayes_classify.sieve b/resources/config/spamfilter/scripts/bayes_classify.sieve index 3467754b..ce5835df 100644 --- a/resources/config/spamfilter/scripts/bayes_classify.sieve +++ b/resources/config/spamfilter/scripts/bayes_classify.sieve @@ -6,7 +6,7 @@ if eval "!t.SPAM_TRAP && !t.TRUSTED_REPLY" { # min_prob_strength: 0.05 # min_learns: 200 - let "bayes_result" "bayes_classify('spamdb/token-lookup', body_and_subject, [2, 11, 0.05, 200])"; + let "bayes_result" "bayes_classify(SPAM_DB, body_and_subject, [2, 11, 0.05, 200])"; if eval "!is_empty(bayes_result)" { if eval "bayes_result > 0.7" { let "t.BAYES_SPAM" "1"; diff --git a/resources/config/spamfilter/scripts/config.sieve b/resources/config/spamfilter/scripts/config.sieve index 2d5d7bbd..45779a55 100644 --- a/resources/config/spamfilter/scripts/config.sieve +++ b/resources/config/spamfilter/scripts/config.sieve @@ -28,6 +28,8 @@ let "SCORE_DISCARD_THRESHOLD" "0"; # Reject messages with a score above this threshold let "SCORE_REJECT_THRESHOLD" "0"; -# Directory name to use for local domain lookups -let "DOMAIN_DIRECTORY" "'default'"; +# Directory name to use for local domain lookups (leave empty for default) +let "DOMAIN_DIRECTORY" ""; +# Store to use for Bayes tokens and ids (leave empty for default) +let "SPAM_DB" ""; diff --git a/resources/config/spamfilter/scripts/dmarc.sieve b/resources/config/spamfilter/scripts/dmarc.sieve index f0368620..49fcb7cd 100644 --- a/resources/config/spamfilter/scripts/dmarc.sieve +++ b/resources/config/spamfilter/scripts/dmarc.sieve @@ -64,13 +64,13 @@ if eval "header.DKIM-Signature.exists" { } # Check allowlists -if eval "lookup('spam/dmarc-allow', from_domain)" { +if eval "key_exists('spam/dmarc-allow', from_domain)" { if eval "t.DMARC_POLICY_ALLOW" { let "t.ALLOWLIST_DMARC" "1"; } else { let "t.BLOCKLIST_DMARC" "1"; } -} elsif eval "lookup('spam/spf-dkim-allow', from_domain)" { +} elsif eval "key_exists('spam/spf-dkim-allow', from_domain)" { let "is_dkim_pass" "contains(env.dkim.domains, from_domain) || t.ARC_ALLOW"; if eval "is_dkim_pass && t.SPF_ALLOW" { diff --git a/resources/config/spamfilter/scripts/epilogue.sieve b/resources/config/spamfilter/scripts/epilogue.sieve index 1967bb84..4eb1fd33 100644 --- a/resources/config/spamfilter/scripts/epilogue.sieve +++ b/resources/config/spamfilter/scripts/epilogue.sieve @@ -2,8 +2,8 @@ # Train the bayes classifier automatically if eval "AUTOLEARN_ENABLE && (score >= AUTOLEARN_SPAM_THRESHOLD || score <= AUTOLEARN_HAM_THRESHOLD)" { let "is_spam" "score >= AUTOLEARN_SPAM_THRESHOLD"; - eval "bayes_is_balanced('spamdb/token-lookup', is_spam, AUTOLEARN_SPAM_HAM_BALANCE) && - bayes_train('spamdb/token-insert', body_and_subject, is_spam)"; + eval "bayes_is_balanced(SPAM_DB, is_spam, AUTOLEARN_SPAM_HAM_BALANCE) && + bayes_train(SPAM_DB, body_and_subject, is_spam)"; } # Process score actions diff --git a/resources/config/spamfilter/scripts/from.sieve b/resources/config/spamfilter/scripts/from.sieve index f22f9e43..f38b7fa9 100644 --- a/resources/config/spamfilter/scripts/from.sieve +++ b/resources/config/spamfilter/scripts/from.sieve @@ -16,9 +16,9 @@ if eval "from_count > 0" { let "t.WWW_DOT_DOMAIN" "1"; } - if eval "lookup('spam/free-domains', from_domain_sld)" { + if eval "key_exists('spam/free-domains', from_domain_sld)" { let "t.FREEMAIL_FROM" "1"; - } elsif eval "lookup('spam/disposable-domains', from_domain_sld)" { + } elsif eval "key_exists('spam/disposable-domains', from_domain_sld)" { let "t.DISPOSABLE_FROM" "1"; } } else { @@ -125,9 +125,9 @@ if eval "!is_empty(envelope.from)" { } if eval "!is_empty(envfrom_domain_sld)" { - if eval "lookup('spam/free-domains', envfrom_domain_sld)" { + if eval "key_exists('spam/free-domains', envfrom_domain_sld)" { let "t.FREEMAIL_ENVFROM" "1"; - } elsif eval "lookup('spam/disposable-domains', envfrom_domain_sld)" { + } elsif eval "key_exists('spam/disposable-domains', envfrom_domain_sld)" { let "t.DISPOSABLE_ENVFROM" "1"; } diff --git a/resources/config/spamfilter/scripts/greylist.sieve b/resources/config/spamfilter/scripts/greylist.sieve index 168fbf69..5b82eea1 100644 --- a/resources/config/spamfilter/scripts/greylist.sieve +++ b/resources/config/spamfilter/scripts/greylist.sieve @@ -1,9 +1,9 @@ -set "triplet" "${env.remote_ip}.${envelope.from}.${envelope.to}"; +set "triplet" "g:${env.remote_ip}.${envelope.from}.${envelope.to}"; -if eval "!lookup('spamdb/id-lookup', triplet)" { +if eval "!key_exists(SPAMDB, triplet)" { # Greylist sender for 30 days - eval "lookup_map('spamdb/id-insert', [triplet, 2592000])"; + eval "key_set(SPAMDB, triplet, '', 2592000)"; reject "422 4.2.2 Greylisted, please try again in a few moments."; stop; } diff --git a/resources/config/spamfilter/scripts/mime.sieve b/resources/config/spamfilter/scripts/mime.sieve index 87c9f9eb..de47e6d7 100644 --- a/resources/config/spamfilter/scripts/mime.sieve +++ b/resources/config/spamfilter/scripts/mime.sieve @@ -186,9 +186,9 @@ foreverypart { } let "name_parts" "rsplit(to_lowercase(attach_name), '.')"; if eval "count(name_parts) > 1" { - let "ext_type" "lookup_map('spam/mime-types', name_parts[0])"; + let "ext_type" "key_get('spam/mime-types', name_parts[0])"; if eval "!is_empty(ext_type)" { - let "ext_type_double" "lookup_map('spam/mime-types', name_parts[1])"; + let "ext_type_double" "key_get('spam/mime-types', name_parts[1])"; if eval "contains(ext_type, 'BAD')" { # Bad extension if eval "contains(ext_type_double, 'BAD')" { diff --git a/resources/config/spamfilter/scripts/rbl.sieve b/resources/config/spamfilter/scripts/rbl.sieve index fc23b95c..7c29812e 100644 --- a/resources/config/spamfilter/scripts/rbl.sieve +++ b/resources/config/spamfilter/scripts/rbl.sieve @@ -175,7 +175,7 @@ while "i < domains_len" { if eval "!contains(domain, '.') || is_ip_addr(domain) || is_local_domain(DOMAIN_DIRECTORY, domain_part(domain, 'sld')) || - lookup('spam/domains-allow', domain)" { + key_exists('spam/domains-allow', domain)" { continue; } @@ -327,7 +327,7 @@ while "i < urls_len" { # Skip URLs pointing to local or trusted domains let "domain" "domain_part(uri_part(url, 'host'), 'sld')"; if eval "is_local_domain(DOMAIN_DIRECTORY, domain) || - lookup('spam/domains-allow', domain)" { + key_exists('spam/domains-allow', domain)" { continue; } diff --git a/resources/config/spamfilter/scripts/recipient.sieve b/resources/config/spamfilter/scripts/recipient.sieve index 60d0dce0..6cf38a37 100644 --- a/resources/config/spamfilter/scripts/recipient.sieve +++ b/resources/config/spamfilter/scripts/recipient.sieve @@ -89,13 +89,13 @@ if eval "rcpt_count > 0" { # Check for freemail or disposable domains let "domain" "domain_part(email_part(addr, 'domain'), 'sld')"; if eval "!is_empty(domain)" { - if eval "lookup('spam/free-domains', domain)" { + if eval "key_exists('spam/free-domains', domain)" { if eval "!t.FREEMAIL_TO && contains_ignore_case(recipients_to, addr)" { let "t.FREEMAIL_TO" "1"; } elsif eval "!t.FREEMAIL_CC && contains_ignore_case(recipients_cc, addr)" { let "t.FREEMAIL_CC" "1"; } - } elsif eval "lookup('spam/disposable-domains', domain)" { + } elsif eval "key_exists('spam/disposable-domains', domain)" { if eval "!t.DISPOSABLE_TO && contains_ignore_case(recipients_to, addr)" { let "t.DISPOSABLE_TO" "1"; } elsif eval "!t.DISPOSABLE_CC && contains_ignore_case(recipients_cc, addr)" { diff --git a/resources/config/spamfilter/scripts/replies_in.sieve b/resources/config/spamfilter/scripts/replies_in.sieve index c3471e55..4f485bde 100644 --- a/resources/config/spamfilter/scripts/replies_in.sieve +++ b/resources/config/spamfilter/scripts/replies_in.sieve @@ -1,4 +1,12 @@ -if eval "lookup('spamdb/id-lookup', header.In-Reply-To:References)" { - let "t.TRUSTED_REPLY" "1"; +let "message_ids" "header.In-Reply-To:References"; + +let "i" "count(message_ids)"; +while "i > 0" { + let "i" "i - 1"; + + if eval "key_exists(SPAM_DB, 'm:' + message_ids[i])" { + let "t.TRUSTED_REPLY" "1"; + break; + } } diff --git a/resources/config/spamfilter/scripts/replies_out.sieve b/resources/config/spamfilter/scripts/replies_out.sieve index 26729c92..76c21875 100644 --- a/resources/config/spamfilter/scripts/replies_out.sieve +++ b/resources/config/spamfilter/scripts/replies_out.sieve @@ -4,9 +4,9 @@ let "message_id" "header.Message-ID"; if eval "!is_empty(message_id)" { # Store the message ID for 30 days - eval "lookup_map('spamdb/id-insert', [message_id, 2592000])"; + eval "key_set(SPAM_DB, 'm:' + message_id, '', 2592000)"; - if eval "AUTOLEARN_ENABLE && AUTOLEARN_REPLIES_HAM && bayes_is_balanced('spamdb/token-lookup', false, AUTOLEARN_SPAM_HAM_BALANCE)" { - eval "bayes_train('spamdb/token-insert', thread_name(header.subject) + ' ' + body.to_text, false)"; + if eval "AUTOLEARN_ENABLE && AUTOLEARN_REPLIES_HAM && bayes_is_balanced(SPAM_DB, false, AUTOLEARN_SPAM_HAM_BALANCE)" { + eval "bayes_train(SPAM_DB, thread_name(header.subject) + ' ' + body.to_text, false)"; } } diff --git a/resources/config/spamfilter/scripts/replyto.sieve b/resources/config/spamfilter/scripts/replyto.sieve index 6d3265c8..d670a09d 100644 --- a/resources/config/spamfilter/scripts/replyto.sieve +++ b/resources/config/spamfilter/scripts/replyto.sieve @@ -49,12 +49,12 @@ if eval "!is_empty(rto_raw)" { let "t.REPLYTO_ADDR_EQ_FROM" "1"; } - if eval "lookup('spam/free-domains', rto_domain_sld)" { + if eval "key_exists('spam/free-domains', rto_domain_sld)" { let "t.FREEMAIL_REPLYTO" "1"; - if eval "rto_domain_sld != from_domain_sld && lookup('spam/free-domains', from_domain_sld)" { + if eval "rto_domain_sld != from_domain_sld && key_exists('spam/free-domains', from_domain_sld)" { let "t.FREEMAIL_REPLYTO_NEQ_FROM_DOM" "1"; } - } elsif eval "lookup('spam/disposable-domains', rto_domain_sld)" { + } elsif eval "key_exists('spam/disposable-domains', rto_domain_sld)" { let "t.DISPOSABLE_REPLYTO" "1"; } diff --git a/resources/config/spamfilter/scripts/reputation.sieve b/resources/config/spamfilter/scripts/reputation.sieve index 28d7d429..71ff4e99 100644 --- a/resources/config/spamfilter/scripts/reputation.sieve +++ b/resources/config/spamfilter/scripts/reputation.sieve @@ -37,15 +37,20 @@ while "i > 0" { let "token_id" "token_ids[i]"; # Lookup reputation - let "token_rep" "lookup_map('spamdb/reputation-lookup', token_id)"; - - # Update reputation - eval "lookup_map('spamdb/reputation-insert', [token_id, score])"; + let "token_rep" "key_get(SPAM_DB, token_id)"; if eval "is_empty(token_rep)" { + # Set reputation + eval "key_set(SPAM_DB, token_id, [score, 1], 2592000)"; continue; } + # Update reputation + let "token_score" "token_rep[0]"; + let "token_count" "token_rep[1]"; + let "updated_score" "(token_count + 1) * (score + 0.98 * token_score) / (0.98 * token_count + 1)"; + eval "key_set(SPAM_DB, token_id, [updated_score, token_count + 1], 2592000)"; + # Assign weight let "weight" ""; if eval "starts_with(token_id, 'f:')" { @@ -64,7 +69,7 @@ while "i > 0" { continue; } - let "reputation" "reputation + (token_rep[0] / token_rep[1] * weight)"; + let "reputation" "reputation + (token_score / token_count * weight)"; } # Adjust score using a 0.5 factor diff --git a/resources/config/spamfilter/scripts/scores.sieve b/resources/config/spamfilter/scripts/scores.sieve index 95a12430..0f8421c1 100644 --- a/resources/config/spamfilter/scripts/scores.sieve +++ b/resources/config/spamfilter/scripts/scores.sieve @@ -5,7 +5,7 @@ let "spam_result" ""; while "i > 0" { let "i" "i - 1"; let "tag" "tags[i]"; - let "tag_score" "lookup_map('spam/scores', tag)"; + let "tag_score" "key_get('spam/scores', tag)"; if eval "is_number(tag_score)" { let "score" "score + tag_score"; diff --git a/resources/config/spamfilter/scripts/spamtrap.sieve b/resources/config/spamfilter/scripts/spamtrap.sieve index 27cae85a..055735f1 100644 --- a/resources/config/spamfilter/scripts/spamtrap.sieve +++ b/resources/config/spamfilter/scripts/spamtrap.sieve @@ -1,7 +1,7 @@ # Check if the message was sent to a spam trap address -if eval "AUTOLEARN_ENABLE && lookup('spam/trap-address', envelope.to)" { - eval "bayes_is_balanced('spamdb/token-lookup', false, AUTOLEARN_SPAM_HAM_BALANCE) && bayes_train('spamdb/token-insert', body_and_subject, true)"; +if eval "AUTOLEARN_ENABLE && key_exists('spam/trap-address', envelope.to)" { + eval "bayes_is_balanced(SPAM_DB, false, AUTOLEARN_SPAM_HAM_BALANCE) && bayes_train(SPAM_DB, body_and_subject, true)"; let "t.SPAM_TRAP" "1"; # Disable autolearn so the classifier is not trained twice diff --git a/resources/config/spamfilter/scripts/url.sieve b/resources/config/spamfilter/scripts/url.sieve index 43a3c1d9..97322883 100644 --- a/resources/config/spamfilter/scripts/url.sieve +++ b/resources/config/spamfilter/scripts/url.sieve @@ -27,13 +27,13 @@ while "i > 0" { let "host_sld" "domain_part(host_lc, 'sld')"; # Skip local and trusted domains - if eval "is_local_domain(DOMAIN_DIRECTORY, host_sld) || lookup('spam/domains-allow', host_sld)" { + if eval "is_local_domain(DOMAIN_DIRECTORY, host_sld) || key_exists('spam/domains-allow', host_sld)" { continue; } if eval "!is_ip && (!t.REDIRECTOR_URL || !t.URL_REDIRECTOR_NESTED) && - lookup('spam/redirectors', host_sld)" { + key_exists('spam/redirectors', host_sld)" { let "t.REDIRECTOR_URL" "1"; let "redir_count" "1"; @@ -48,7 +48,7 @@ while "i > 0" { let "host_lc" "to_lowercase(host)"; let "host_sld" "domain_part(host_lc, 'sld')"; - if eval "!is_ip && lookup('spam/redirectors', host_sld)" { + if eval "!is_ip && key_exists('spam/redirectors', host_sld)" { let "redir_count" "redir_count + 1"; } else { break; @@ -110,10 +110,10 @@ while "i > 0" { } # Phishing checks (refresh OpenPhish every 12 hours, PhishTank every 6 hours) - if eval "lookup_remote('https://openphish.com/feed.txt', url, [43200, 'list'])" { + if eval "key_exists_http('https://openphish.com/feed.txt', url, [43200, 'list'])" { let "t.PHISHED_OPENPHISH" "1"; } - if eval "lookup_remote('http://data.phishtank.com/data/online-valid.csv', url, [21600, 'csv', 1, ',', true])" { + if eval "key_exists_http('http://data.phishtank.com/data/online-valid.csv', url, [21600, 'csv', 1, ',', true])" { let "t.PHISHED_PHISHTANK" "1"; } diff --git a/tests/src/smtp/inbound/antispam.rs b/tests/src/smtp/inbound/antispam.rs index 2471ad5d..d5be8012 100644 --- a/tests/src/smtp/inbound/antispam.rs +++ b/tests/src/smtp/inbound/antispam.rs @@ -46,25 +46,6 @@ duplicate-expiry = "7d" type = "sqlite" path = "%PATH%/test_antispam.db" -[store."spamdb".pool] -max-connections = 10 -min-connections = 0 -idle-timeout = "5m" - -[store."spamdb".query] -token-insert = "INSERT INTO bayes_tokens (h1, h2, ws, wh) VALUES (?, ?, ?, ?) - ON CONFLICT(h1, h2) - DO UPDATE SET ws = ws + excluded.ws, wh = wh + excluded.wh" -token-lookup = "SELECT ws, wh FROM bayes_tokens WHERE h1 = ? AND h2 = ?" -id-insert = "INSERT INTO seen_ids (id, ttl) VALUES (?, datetime('now', ? || ' seconds'))" -id-lookup = "SELECT 1 FROM seen_ids WHERE id = ? AND ttl > CURRENT_TIMESTAMP" -id-cleanup = "DELETE FROM seen_ids WHERE ttl < CURRENT_TIMESTAMP" -reputation-insert = "INSERT INTO reputation (token, score, count, ttl) VALUES (?, ?, 1, datetime('now', '30 days')) - ON CONFLICT(token) - DO UPDATE SET score = (count + 1) * (excluded.score + 0.98 * score) / (0.98 * count + 1), count = count + 1, ttl = excluded.ttl" -reputation-lookup = "SELECT score, count FROM reputation WHERE token = ?" -reputation-cleanup = "DELETE FROM reputation WHERE ttl < CURRENT_TIMESTAMP" - [store."default"] type = "memory" @@ -130,26 +111,6 @@ public-suffix = "file://%LIST_PATH%/public-suffix.dat" [sieve.trusted.scripts] "#; -const CREATE_TABLES: &[&str; 3] = &[ - "CREATE TABLE IF NOT EXISTS bayes_tokens ( -h1 INTEGER NOT NULL, -h2 INTEGER NOT NULL, -ws INTEGER, -wh INTEGER, -PRIMARY KEY (h1, h2) -)", - "CREATE TABLE IF NOT EXISTS seen_ids ( - id STRING NOT NULL PRIMARY KEY, - ttl DATETIME NOT NULL -)", - "CREATE TABLE IF NOT EXISTS reputation ( -token STRING NOT NULL PRIMARY KEY, -score FLOAT NOT NULL DEFAULT '0', -count INT(11) NOT NULL DEFAULT '0', -ttl DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP -)", -]; - #[tokio::test(flavor = "multi_thread")] async fn antispam() { /*tracing::subscriber::set_global_default( @@ -259,12 +220,6 @@ async fn antispam() { let config = &mut core.session.config; config.rcpt.relay = IfBlock::new(true); - // Create tables - let sdb = ctx.stores.lookup_stores.get("spamdb").unwrap(); - for query in CREATE_TABLES { - sdb.query::(query, vec![]).await.expect(query); - } - // Add mock DNS entries for (domain, ip) in [ ("bank.com", "127.0.0.1"), diff --git a/tests/src/smtp/mod.rs b/tests/src/smtp/mod.rs index 0c932112..c0ca8b93 100644 --- a/tests/src/smtp/mod.rs +++ b/tests/src/smtp/mod.rs @@ -48,7 +48,7 @@ use smtp::{ }, core::{ throttle::ThrottleKeyHasherBuilder, QueueCore, ReportCore, Resolvers, SessionCore, - SieveConfig, SieveCore, TlsConnectors, SMTP, + SieveCore, TlsConnectors, SMTP, }, outbound::dane::DnssecResolver, }; @@ -434,14 +434,12 @@ impl TestConfig for SieveCore { runtime: Runtime::new_with_context(SieveContext::default()), scripts: AHashMap::new(), lookup: AHashMap::new(), - config: SieveConfig { - from_addr: "MAILER-DAEMON@example.org".to_string(), - from_name: "Mailer Daemon".to_string(), - return_path: "".to_string(), - sign: vec![], - directories: Default::default(), - lookup_stores: Default::default(), - }, + from_addr: "MAILER-DAEMON@example.org".to_string(), + from_name: "Mailer Daemon".to_string(), + return_path: "".to_string(), + sign: vec![], + directories: Default::default(), + lookup_stores: Default::default(), } } } diff --git a/tests/src/store/blob.rs b/tests/src/store/blob.rs index 4feb5bb8..7bd80cfe 100644 --- a/tests/src/store/blob.rs +++ b/tests/src/store/blob.rs @@ -111,7 +111,7 @@ pub async fn blob_tests() { ); // Purge expired blobs - store.blob_hash_purge(blob_store.clone()).await.unwrap(); + store.purge_blobs(blob_store.clone()).await.unwrap(); // Blob hash should no longer exist assert!(!store.blob_hash_exists(&hash).await.unwrap()); @@ -192,7 +192,7 @@ pub async fn blob_tests() { ); // Purge expired blobs and make sure nothing else is deleted - store.blob_hash_purge(blob_store.clone()).await.unwrap(); + store.purge_blobs(blob_store.clone()).await.unwrap(); for (pos, (blob, blob_class)) in [ (b"abc", BlobClass::Reserved { account_id: 0 }), ( @@ -266,7 +266,7 @@ pub async fn blob_tests() { .unwrap(); // Purge and make sure blob is deleted - store.blob_hash_purge(blob_store.clone()).await.unwrap(); + store.purge_blobs(blob_store.clone()).await.unwrap(); for (pos, (blob, blob_class)) in [ ( b"789", @@ -314,7 +314,7 @@ pub async fn blob_tests() { // Unlink all blobs from accountId 1 and purge store.blob_hash_unlink_account(1).await.unwrap(); - store.blob_hash_purge(blob_store.clone()).await.unwrap(); + store.purge_blobs(blob_store.clone()).await.unwrap(); // Make sure only accountId 0's blobs are left for (pos, (blob, blob_class)) in [