Compare commits
15
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
f398d95062 | ||
|
|
81ec917d74 | ||
|
|
e2ab26ad19 | ||
|
|
0b4aa9c084 | ||
|
|
4e6c8b916e | ||
|
|
7735780807 | ||
|
|
e223f7d327 | ||
|
|
30df055e39 | ||
|
|
5393c4405a | ||
|
|
cc532b914c | ||
|
|
c09eff2214 | ||
|
|
eba4c7a32e | ||
|
|
17426f6d60 | ||
|
|
d7c9416713 | ||
|
|
238079da66 |
+8
-2
@@ -1,10 +1,16 @@
|
||||
// Ignore everything
|
||||
# Ignore everything
|
||||
*
|
||||
|
||||
// Allow what is needed
|
||||
# Allow what is needed
|
||||
!crates
|
||||
!tests
|
||||
!resources
|
||||
|
||||
# The patched dependency Cargo.toml's [patch.crates-io] points at. Without
|
||||
# it the build context has no vendor/, and `cargo chef cook` fails on
|
||||
# "failed to load source for dependency sieve-rs" -- which CI cannot see,
|
||||
# because CI builds from a checkout and only the image build has a context.
|
||||
!vendor
|
||||
|
||||
!Cargo.lock
|
||||
!Cargo.toml
|
||||
|
||||
@@ -35,6 +35,11 @@ jobs:
|
||||
- run: python3 tools/fork/name-check.py
|
||||
- if: always()
|
||||
run: python3 tools/fork/notice-check.py
|
||||
# Cargo can patch a dependency to a directory in this repository, and
|
||||
# the image builds from a context .dockerignore prunes to almost
|
||||
# nothing. CI never sees the difference; a release does.
|
||||
- if: always()
|
||||
run: python3 tools/fork/context-check.py
|
||||
|
||||
build:
|
||||
# Either runner (host1 or host2): the build needs no docker socket.
|
||||
|
||||
@@ -14,8 +14,11 @@
|
||||
# crates/types/src/branding.rs, not Cargo.toml, and the image is tagged
|
||||
# with it, so a tag beside an unbumped macro would publish an image that
|
||||
# reports a different version from its tag.
|
||||
# * the tag must be on main, so an image never describes code that was never
|
||||
# reviewed onto the default branch.
|
||||
# * the tag must be on main or on a release/* branch, so an image never
|
||||
# describes code that was never reviewed onto one of them. A release/*
|
||||
# branch carries a hotfix: it starts at an earlier release tag, takes
|
||||
# fixes through pull requests into it, and is tagged there, so production
|
||||
# can get a fix without everything that has landed on main since.
|
||||
#
|
||||
# :latest moves with every published tag: tags are cut by the weekly release
|
||||
# (or by hand for a real release); there are no prerelease tags here.
|
||||
@@ -57,8 +60,13 @@ jobs:
|
||||
echo "Refusing to publish an image that would report the wrong version." >&2
|
||||
exit 1
|
||||
fi
|
||||
git merge-base --is-ancestor "$(git rev-parse "${TAG}^{commit}")" origin/main \
|
||||
|| { echo "$TAG is not on main" >&2; exit 1; }
|
||||
commit="$(git rev-parse "${TAG}^{commit}")"
|
||||
on=""
|
||||
for ref in origin/main $(git for-each-ref --format='%(refname:short)' 'refs/remotes/origin/release/*'); do
|
||||
if git merge-base --is-ancestor "$commit" "$ref"; then on="$ref"; break; fi
|
||||
done
|
||||
[ -n "$on" ] || { echo "$TAG is not on main or a release/* branch" >&2; exit 1; }
|
||||
echo "$TAG is on $on"
|
||||
echo "version=$V" >> "$GITHUB_OUTPUT"
|
||||
echo "version $V"
|
||||
|
||||
|
||||
@@ -12,6 +12,9 @@
|
||||
# An issue is opened once per release: an existing one with the same title,
|
||||
# open or closed, stops a second.
|
||||
#
|
||||
# It also watches spam-filter, whose rules the server bundles
|
||||
# (resources/spam-filter/), and opens an issue for a newer release.
|
||||
#
|
||||
# Daily 06:17 UTC; run it by hand with workflow_dispatch.
|
||||
name: upstream-watch
|
||||
|
||||
@@ -64,7 +67,7 @@ jobs:
|
||||
and key(r["tag_name"]) > key(base)),
|
||||
key=lambda r: key(r["tag_name"]))
|
||||
if not newer:
|
||||
print(f"Up to date: {base} is the newest upstream release."); sys.exit(0)
|
||||
print(f"Up to date: {base} is the newest upstream release.")
|
||||
|
||||
# Titles and bodies stay free of the upstream project's name, as the
|
||||
# rest of the fork's user-visible text does.
|
||||
@@ -85,4 +88,35 @@ jobs:
|
||||
"add any new third-party notices to `THIRD-PARTY.md`, then merge `upstream` into `main`.")
|
||||
issue = call("POST", f"{api}/issues", {"title": title, "body": body})
|
||||
print(f"{tag}: opened #{issue['number']}.")
|
||||
|
||||
# The spam filter rules bundled with the server (resources/spam-filter/):
|
||||
# an issue when spam-filter publishes a newer release than the one
|
||||
# BUNDLED_SPAM_RULES_VERSION names on main.
|
||||
src = call("GET", f"{api}/contents/crates/common/src/manager/spam_rules.rs?ref=main")
|
||||
import base64
|
||||
text = base64.b64decode(src["content"]).decode()
|
||||
m = re.search(r'BUNDLED_SPAM_RULES_VERSION: &str = "(\d+\.\d+\.\d+)"', text)
|
||||
if not m:
|
||||
print("Can't read BUNDLED_SPAM_RULES_VERSION from spam_rules.rs", file=sys.stderr); sys.exit(1)
|
||||
bundled = "v" + m.group(1)
|
||||
rels = call("GET", "https://api.github.com/repos/stalwartlabs/spam-filter/releases?per_page=30", token=None)
|
||||
newer = sorted((r for r in rels
|
||||
if not r["draft"] and not r["prerelease"] and SEMVER.match(r["tag_name"])
|
||||
and key(r["tag_name"]) > key(bundled)),
|
||||
key=lambda r: key(r["tag_name"]))
|
||||
if not newer:
|
||||
print(f"Up to date: the bundled spam rules are {bundled}, the newest release."); sys.exit(0)
|
||||
latest = newer[-1]
|
||||
tag = latest["tag_name"]
|
||||
title = f"Update the bundled spam rules to {tag}"
|
||||
existing = {i["title"] for i in call("GET", f"{api}/issues?state=all&type=issues&q=bundled+spam+rules&limit=50")}
|
||||
if title in existing:
|
||||
print(f"spam rules {tag}: issue already exists."); sys.exit(0)
|
||||
body = (f"spam-filter published {tag} on {latest['published_at'][:10]}. "
|
||||
f"The server bundles {bundled}.\n\n"
|
||||
"Update it as resources/spam-filter/README.md describes: take the rules file "
|
||||
f"from the {tag} release (by tag, not `latest`), set BUNDLED_SPAM_RULES_VERSION, "
|
||||
"and run the antispam test.")
|
||||
issue = call("POST", f"{api}/issues", {"title": title, "body": body})
|
||||
print(f"spam rules {tag}: opened #{issue['number']}.")
|
||||
PY
|
||||
|
||||
@@ -19,6 +19,10 @@ RUN export DEBIAN_FRONTEND=noninteractive && \
|
||||
g++-x86-64-linux-gnu binutils-x86-64-linux-gnu
|
||||
RUN rustup target add "$(cat /target.txt)"
|
||||
COPY --from=planner /recipe.json /recipe.json
|
||||
# inbuxa: [patch.crates-io] points sieve-rs at vendor/, and the recipe only
|
||||
# carries the workspace's own manifests, so cooking the dependencies needs the
|
||||
# vendored crate itself (the context allows it since #27; this puts it here).
|
||||
COPY vendor/ vendor/
|
||||
RUN RUSTFLAGS="$(cat /flags.txt)" cargo chef cook --target "$(cat /target.txt)" --release --no-default-features --features "sqlite postgres mysql rocks s3 redis azure nats" --recipe-path /recipe.json
|
||||
COPY . .
|
||||
RUN RUSTFLAGS="$(cat /flags.txt)" cargo build --target "$(cat /target.txt)" --release -p inbuxa --no-default-features --features "sqlite postgres mysql rocks s3 redis azure nats"
|
||||
|
||||
+1
-1
@@ -24,7 +24,7 @@ carry their own license files.
|
||||
| `crates/common/src/network/acme/directory.rs`, `crates/common/src/network/acme/jose.rs`, `crates/common/src/network/acme/order.rs` | [rustls-acme](https://github.com/FlorianUekermann/rustls-acme) (MIT or Apache-2.0) | Copyright (c) Florian Uekermann |
|
||||
| `crates/types/src/id.rs` | [crockford](https://github.com/archer884/crockford) (MIT or Apache-2.0) | Copyright (c) 2017 J/A <archer884@gmail.com> |
|
||||
| `crates/nlp/src/tokenizers/types.rs` | test cases from [linkify](https://github.com/robinst/linkify) (MIT or Apache-2.0) | Copyright (c) 2017 Robin Stocker |
|
||||
| `tests/resources/smtp/antispam/spam-filter-rules.json.gz` | the published rules of [spam-filter](https://github.com/stalwartlabs/spam-filter) v3.0.2, unmodified, for the spam filter's tests (MIT or Apache-2.0) | Copyright (C) 2024, Stalwart Labs LLC |
|
||||
| `resources/spam-filter/spam-filter-rules.json.gz` | the published rules of [spam-filter](https://github.com/stalwartlabs/spam-filter) v3.0.2, unmodified, built into the server as its default spam rules (MIT or Apache-2.0) | Copyright (C) 2024, Stalwart Labs LLC |
|
||||
|
||||
Each notice above applies with this permission notice:
|
||||
|
||||
|
||||
@@ -243,7 +243,8 @@ impl SpamFilterConfig {
|
||||
spam_threshold: spam.score_spam.into_inner() as f32,
|
||||
},
|
||||
grey_list_expiry: spam.greylist_for.map(|d| d.into_inner().as_secs()),
|
||||
spam_rules_url: spam.spam_filter_rules_url,
|
||||
// inbuxa: unset, empty or upstream's old default means the bundled rules
|
||||
spam_rules_url: crate::manager::spam_rules::rules_url(spam.spam_filter_rules_url),
|
||||
url_client: utils::http::http_client_builder(true)
|
||||
.pool_max_idle_per_host(0)
|
||||
.redirect(reqwest::redirect::Policy::none())
|
||||
|
||||
@@ -530,13 +530,22 @@ async fn insert_safe_defaults(bp: &mut Bootstrap) -> trc::Result<()> {
|
||||
use store::write::BatchBuilder;
|
||||
use types::id::Id;
|
||||
|
||||
if bp.registry.count_object(ObjectType::SpamRule).await? == 0
|
||||
&& bp
|
||||
.registry
|
||||
// inbuxa: rules are always to hand, since a copy ships with the server
|
||||
// (spam_rules). They load on first boot, and again when the bundled
|
||||
// version differs from the one last loaded, which only adds what's
|
||||
// missing: new tags and rules, never a changed score.
|
||||
let rules_url = super::spam_rules::rules_url(
|
||||
bp.registry
|
||||
.object::<SpamSettings>(Id::singleton())
|
||||
.await?
|
||||
.is_none_or(|spam| spam.spam_filter_rules_url.is_some())
|
||||
{
|
||||
.and_then(|spam| spam.spam_filter_rules_url),
|
||||
);
|
||||
let bundled_is_new = rules_url.is_none()
|
||||
&& super::spam_rules::applied_version(&bp.data_store)
|
||||
.await?
|
||||
.as_deref()
|
||||
!= Some(super::spam_rules::BUNDLED_SPAM_RULES_VERSION);
|
||||
if bp.registry.count_object(ObjectType::SpamRule).await? == 0 || bundled_is_new {
|
||||
let mut batch = BatchBuilder::new();
|
||||
batch.schedule_task(Task::SpamFilterMaintenance(TaskSpamFilterMaintenance {
|
||||
maintenance_type: TaskSpamFilterMaintenanceType::UpdateRules,
|
||||
|
||||
@@ -22,6 +22,7 @@ pub mod console;
|
||||
pub mod defaults;
|
||||
pub mod first_party;
|
||||
pub mod restore;
|
||||
pub mod spam_rules; // inbuxa: rules bundled with the server
|
||||
|
||||
pub const SPAM_TRAINER_KEY: &[u8] = "INBUXA_SPAM_TRAIN_DATA.lz4".as_bytes();
|
||||
pub const SPAM_CLASSIFIER_KEY: &[u8] = "INBUXA_SPAM_CLASSIFIER_MODEL.lz4".as_bytes();
|
||||
|
||||
@@ -0,0 +1,103 @@
|
||||
/*
|
||||
* SPDX-FileCopyrightText: 2026 Coffey Labs
|
||||
*
|
||||
* SPDX-License-Identifier: AGPL-3.0-only
|
||||
*/
|
||||
|
||||
//! inbuxa: the spam filter rules that ship with the server.
|
||||
//!
|
||||
//! Upstream fetches its latest published rules from GitHub at run time, so
|
||||
//! scoring changes with a release nobody here tested and depends on reaching
|
||||
//! it. The fork embeds a pinned copy (resources/spam-filter/, with its version
|
||||
//! and license) and uses it whenever no other source is configured. The rules
|
||||
//! URL remains an operator override (`https://` or `file://`).
|
||||
//!
|
||||
//! Loading rules only ever adds what's missing, never changes an existing rule
|
||||
//! or score. They load on first boot, and again whenever the bundled version
|
||||
//! differs from the one last applied, so an upgrade brings new tags (the AI
|
||||
//! classifier's `LLM_*` scores, say) to an install that already had rules.
|
||||
|
||||
use std::io::Read;
|
||||
use store::{
|
||||
SUBSPACE_INBUXA, Store, ValueKey,
|
||||
write::{AnyClass, BatchBuilder, ValueClass},
|
||||
};
|
||||
use trc::AddContext;
|
||||
|
||||
/// The version of spam-filter the embedded rules come from.
|
||||
pub const BUNDLED_SPAM_RULES_VERSION: &str = "3.0.2";
|
||||
|
||||
static BUNDLED_SPAM_RULES: &[u8] =
|
||||
include_bytes!("../../../../resources/spam-filter/spam-filter-rules.json.gz");
|
||||
|
||||
/// Upstream's default rules source, the value every install created before
|
||||
/// the rules were bundled has saved. Read only to treat it as unset.
|
||||
const LEGACY_DEFAULT_URL: &str =
|
||||
"https://github.com/stalwartlabs/spam-filter/releases/latest/download/spam-filter-rules.json.gz";
|
||||
|
||||
/// The URL to fetch rules from, or `None` for the bundled rules. An empty
|
||||
/// setting and upstream's old default both mean the bundled rules.
|
||||
pub fn rules_url(configured: Option<String>) -> Option<String> {
|
||||
configured.filter(|url| !url.trim().is_empty() && url != LEGACY_DEFAULT_URL)
|
||||
}
|
||||
|
||||
/// The bundled rules, uncompressed: the same JSON the rules URL serves.
|
||||
pub fn bundled_rules() -> Result<Vec<u8>, String> {
|
||||
let mut json = Vec::new();
|
||||
mail_auth::flate2::read::GzDecoder::new(BUNDLED_SPAM_RULES)
|
||||
.read_to_end(&mut json)
|
||||
.map_err(|err| format!("Failed to decompress the bundled spam rules: {err}"))?;
|
||||
Ok(json)
|
||||
}
|
||||
|
||||
fn applied_key() -> ValueClass {
|
||||
ValueClass::Any(AnyClass {
|
||||
subspace: SUBSPACE_INBUXA,
|
||||
key: b"Sr".to_vec(),
|
||||
})
|
||||
}
|
||||
|
||||
/// The bundled version last loaded into the registry, if any.
|
||||
pub async fn applied_version(data: &Store) -> trc::Result<Option<String>> {
|
||||
data.get_value::<String>(ValueKey::from(applied_key()))
|
||||
.await
|
||||
.caused_by(trc::location!())
|
||||
}
|
||||
|
||||
/// Records that the bundled rules of this version have been loaded.
|
||||
pub async fn set_applied_version(data: &Store, version: &str) -> trc::Result<()> {
|
||||
let mut batch = BatchBuilder::new();
|
||||
batch.set(applied_key(), version.as_bytes().to_vec());
|
||||
data.write(batch.build_all())
|
||||
.await
|
||||
.caused_by(trc::location!())
|
||||
.map(|_| ())
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
#[test]
|
||||
fn upstream_default_and_empty_mean_bundled() {
|
||||
assert_eq!(rules_url(None), None);
|
||||
assert_eq!(rules_url(Some(String::new())), None);
|
||||
assert_eq!(rules_url(Some(" ".into())), None);
|
||||
assert_eq!(rules_url(Some(LEGACY_DEFAULT_URL.into())), None);
|
||||
assert_eq!(
|
||||
rules_url(Some("file:///srv/rules.json.gz".into())).as_deref(),
|
||||
Some("file:///srv/rules.json.gz")
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn bundled_rules_parse_and_score_the_ai_tags() {
|
||||
let rules: serde_json::Value = serde_json::from_slice(&bundled_rules().unwrap()).unwrap();
|
||||
let tags = rules["SpamTag"].as_array().unwrap();
|
||||
for (tag, score) in [("LLM_UNSOLICITED_HIGH", 3.0), ("LLM_LEGITIMATE_HIGH", -3.0)] {
|
||||
let found = tags.iter().find(|t| t["tag"] == tag).unwrap();
|
||||
assert_eq!(found["score"].as_f64(), Some(score), "{tag}");
|
||||
}
|
||||
assert!(!rules["SpamRule"].as_array().unwrap().is_empty());
|
||||
}
|
||||
}
|
||||
@@ -2,6 +2,8 @@
|
||||
* SPDX-FileCopyrightText: 2020 Stalwart Labs LLC <[email protected]>
|
||||
*
|
||||
* SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL
|
||||
*
|
||||
* Modified by Coffey Labs in 2026 for INBUXA.
|
||||
*/
|
||||
|
||||
use crate::{
|
||||
@@ -15,22 +17,42 @@ use jmap_proto::{error::set::SetError, types::state::State};
|
||||
use jmap_tools::{Key, Value};
|
||||
use registry::{
|
||||
jmap::IntoValue,
|
||||
schema::prelude::{Object, ObjectInner, ObjectType, Property},
|
||||
schema::{
|
||||
prelude::{Object, ObjectInner, ObjectType, Property},
|
||||
structs::Task,
|
||||
},
|
||||
types::{EnumImpl, datetime::UTCDateTime},
|
||||
};
|
||||
use services::task_manager::lock::TaskLockManager;
|
||||
use smtp::reporting::index::{ExternalReportIndex, InternalReportIndex};
|
||||
use std::str::FromStr;
|
||||
use store::{
|
||||
U64_LEN, ValueKey,
|
||||
registry::{RegistryFilter, RegistryFilterValue, RegistryQuery},
|
||||
write::{BatchBuilder, RegistryClass, ValueClass, key::KeySerializer},
|
||||
write::{BatchBuilder, RegistryClass, TaskQueueClass, ValueClass, key::KeySerializer},
|
||||
};
|
||||
use trc::AddContext;
|
||||
use types::id::Id;
|
||||
|
||||
pub(crate) async fn report_set(
|
||||
mut set: RegistrySetResponse<'_>,
|
||||
set: RegistrySetResponse<'_>,
|
||||
) -> trc::Result<RegistrySetResponse<'_>> {
|
||||
// inbuxa: task locks taken to reschedule reports are released however
|
||||
// the request ends; a held lock is renewed, so a leaked one would keep
|
||||
// the report's task from ever running
|
||||
let server = set.server;
|
||||
let mut locked_tasks = Vec::new();
|
||||
let result = report_set_locked(set, &mut locked_tasks).await;
|
||||
for task_id in locked_tasks {
|
||||
server.remove_index_lock(task_id).await;
|
||||
}
|
||||
result
|
||||
}
|
||||
|
||||
async fn report_set_locked<'x>(
|
||||
mut set: RegistrySetResponse<'x>,
|
||||
locked_tasks: &mut Vec<u64>,
|
||||
) -> trc::Result<RegistrySetResponse<'x>> {
|
||||
let object_id = set.object_type.to_id();
|
||||
|
||||
// Reports cannot be created
|
||||
@@ -89,12 +111,45 @@ pub(crate) async fn report_set(
|
||||
.get_value::<Object>(ValueKey::from(key.clone()))
|
||||
.await?
|
||||
{
|
||||
// inbuxa: the report's task shares its id. Hold the task
|
||||
// while its queue rows move, as x:Task/set does, and move the
|
||||
// row the task is actually queued under
|
||||
if !set.server.try_lock_task(item_id).await {
|
||||
set.response.not_updated.append(
|
||||
id,
|
||||
SetError::forbidden().with_description(
|
||||
"The report is being sent and cannot be rescheduled".to_string(),
|
||||
),
|
||||
);
|
||||
continue;
|
||||
}
|
||||
locked_tasks.push(item_id);
|
||||
let queued = set
|
||||
.server
|
||||
.store()
|
||||
.get_value::<Task>(ValueKey::from(ValueClass::TaskQueue(
|
||||
TaskQueueClass::Task { id: item_id },
|
||||
)))
|
||||
.await?;
|
||||
|
||||
match &mut report_obj.inner {
|
||||
ObjectInner::DmarcInternalReport(report) => {
|
||||
report.reschedule_ops(&mut batch, item_id, report_obj.revision, deliver_at);
|
||||
report.reschedule_ops(
|
||||
&mut batch,
|
||||
item_id,
|
||||
report_obj.revision,
|
||||
deliver_at,
|
||||
queued.as_ref(),
|
||||
);
|
||||
}
|
||||
ObjectInner::TlsInternalReport(report) => {
|
||||
report.reschedule_ops(&mut batch, item_id, report_obj.revision, deliver_at);
|
||||
report.reschedule_ops(
|
||||
&mut batch,
|
||||
item_id,
|
||||
report_obj.revision,
|
||||
deliver_at,
|
||||
queued.as_ref(),
|
||||
);
|
||||
}
|
||||
_ => {}
|
||||
}
|
||||
@@ -156,6 +211,9 @@ pub(crate) async fn report_set(
|
||||
.write(batch.build_all())
|
||||
.await
|
||||
.caused_by(trc::location!())?;
|
||||
// inbuxa: a rescheduled report may now be due sooner than the task
|
||||
// manager's next scan
|
||||
set.server.notify_task_queue();
|
||||
}
|
||||
|
||||
Ok(set)
|
||||
|
||||
@@ -463,15 +463,10 @@ pub(crate) async fn task_query(
|
||||
.set_values(typ.is_some()),
|
||||
|key, value| {
|
||||
if let Some(typ) = typ {
|
||||
let task_type =
|
||||
TaskType::from_id(value.deserialize_be_u16(0)?).ok_or_else(|| {
|
||||
trc::StoreEvent::DataCorruption
|
||||
.into_err()
|
||||
.ctx(trc::Key::Key, key.to_vec())
|
||||
.ctx(trc::Key::Value, value.to_vec())
|
||||
.caused_by(trc::location!())
|
||||
})?;
|
||||
if task_type != typ {
|
||||
// inbuxa: a row whose type can't be read matches no type
|
||||
// filter; the task manager logs and repairs it
|
||||
let task_type = value.deserialize_be_u16(0).ok().and_then(TaskType::from_id);
|
||||
if task_type != Some(typ) {
|
||||
return Ok(true);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -40215,7 +40215,7 @@ impl Default for SpamSettings {
|
||||
score_reject: Float::new(0.0f64),
|
||||
score_spam: Float::new(5.0f64),
|
||||
trust_replies: true,
|
||||
spam_filter_rules_url: Some("https://github.com/stalwartlabs/spam-filter/releases/latest/download/spam-filter-rules.json.gz".to_string()),
|
||||
spam_filter_rules_url: None,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -2,6 +2,8 @@
|
||||
* SPDX-FileCopyrightText: 2020 Stalwart Labs LLC <[email protected]>
|
||||
*
|
||||
* SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL
|
||||
*
|
||||
* Modified by Coffey Labs in 2026 for INBUXA.
|
||||
*/
|
||||
|
||||
use crate::task_manager::acme::AcmeTask;
|
||||
@@ -27,6 +29,7 @@ use common::network::limiter::ConcurrencyLimiter;
|
||||
use common::network::{ServerInstance, TcpAcceptor};
|
||||
use common::{Inner, Server};
|
||||
use registry::schema::enums::TaskType;
|
||||
use registry::schema::prelude::ObjectType;
|
||||
use registry::schema::structs::{
|
||||
Task, TaskManager, TaskRetryStrategy, TaskStatus, TaskStatusFailed, TaskStatusRetry,
|
||||
};
|
||||
@@ -337,6 +340,7 @@ impl TaskQueueManager for Server {
|
||||
|
||||
// Retrieve tasks pending to be processed
|
||||
let mut tasks = Vec::new();
|
||||
let mut unreadable = Vec::new();
|
||||
let now = Instant::now();
|
||||
let mut next_event = None;
|
||||
let roles = &self.core.network.roles;
|
||||
@@ -351,12 +355,21 @@ impl TaskQueueManager for Server {
|
||||
let task_id = key.deserialize_be_u64(U64_LEN)?;
|
||||
|
||||
if task_due <= now_timestamp {
|
||||
let task_type_idx = value.deserialize_be_u16(0)?;
|
||||
let task_type = TaskType::from_id(task_type_idx).ok_or_else(|| {
|
||||
trc::StoreEvent::DataCorruption
|
||||
.caused_by(trc::location!())
|
||||
.ctx(trc::Key::Value, value)
|
||||
})?;
|
||||
// inbuxa: a row whose task type can't be read is
|
||||
// set aside, not allowed to end the scan: every
|
||||
// task due after it would wait behind it
|
||||
let Some((task_type_idx, task_type)) = value
|
||||
.deserialize_be_u16(0)
|
||||
.ok()
|
||||
.and_then(|idx| TaskType::from_id(idx).map(|typ| (idx, typ)))
|
||||
else {
|
||||
unreadable.push(UnreadableDueRow {
|
||||
due: task_due,
|
||||
id: task_id,
|
||||
value: value.to_vec(),
|
||||
});
|
||||
return Ok(true);
|
||||
};
|
||||
let enabled = match task_type {
|
||||
TaskType::IndexDocument
|
||||
| TaskType::UnindexDocument
|
||||
@@ -446,6 +459,11 @@ impl TaskQueueManager for Server {
|
||||
);
|
||||
});
|
||||
|
||||
if !unreadable.is_empty() && repair_due_rows(self, unreadable).await {
|
||||
// Look again at once for the rows that were rewritten
|
||||
self.notify_task_queue();
|
||||
}
|
||||
|
||||
if !tasks.is_empty() {
|
||||
trc::event!(
|
||||
TaskManager(TaskManagerEvent::TaskAcquired),
|
||||
@@ -686,3 +704,114 @@ impl TaskResult {
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
/// inbuxa: a task queue row whose task type could not be read.
|
||||
struct UnreadableDueRow {
|
||||
due: u64,
|
||||
id: u64,
|
||||
value: Vec<u8>,
|
||||
}
|
||||
|
||||
/// inbuxa: logs each unreadable queue row and repairs it from the task it
|
||||
/// schedules. The task row says what the task is, so the queue row is
|
||||
/// rewritten with that task's type; a row with no task behind it is removed.
|
||||
///
|
||||
/// Rescheduling an internal DMARC or TLS report wrote the report's object
|
||||
/// type into the queue row instead of the task type. Such a row is the time
|
||||
/// an administrator chose, so the task is moved to it as the reschedule
|
||||
/// meant to do: the task row takes that due, and a queue row left at the
|
||||
/// task's previous due is removed. Returns whether any row was repaired.
|
||||
async fn repair_due_rows(server: &Server, rows: Vec<UnreadableDueRow>) -> bool {
|
||||
let mut repaired = false;
|
||||
for row in rows {
|
||||
let UnreadableDueRow { due, id, value } = row;
|
||||
trc::error!(
|
||||
trc::StoreEvent::DataCorruption
|
||||
.into_err()
|
||||
.id(id)
|
||||
.ctx(trc::Key::Due, trc::Value::Timestamp(due))
|
||||
.ctx(
|
||||
trc::Key::Key,
|
||||
[due.to_be_bytes(), id.to_be_bytes()].concat()
|
||||
)
|
||||
.ctx(trc::Key::Value, value.clone())
|
||||
.details("Unreadable task queue row skipped")
|
||||
.caused_by(trc::location!())
|
||||
);
|
||||
|
||||
let task_key = ValueClass::TaskQueue(TaskQueueClass::Task { id });
|
||||
let due_key = ValueClass::TaskQueue(TaskQueueClass::Due { id, due });
|
||||
let task = match server
|
||||
.store()
|
||||
.get_value::<Task>(ValueKey::from(task_key.clone()))
|
||||
.await
|
||||
{
|
||||
Ok(task) => task,
|
||||
Err(err) => {
|
||||
trc::error!(
|
||||
err.id(id)
|
||||
.details("Failed to read the task of an unreadable queue row.")
|
||||
.caused_by(trc::location!())
|
||||
);
|
||||
continue;
|
||||
}
|
||||
};
|
||||
|
||||
let mut batch = BatchBuilder::new();
|
||||
let action = if let Some(mut task) = task {
|
||||
let task_type = task.object_type();
|
||||
batch.assert_value(task_key.clone(), AssertValue::Some);
|
||||
if rescheduled_report_type(&value) == Some(task_type) {
|
||||
let old_due = task.due_timestamp();
|
||||
if old_due != due {
|
||||
batch.clear(ValueClass::TaskQueue(TaskQueueClass::Due {
|
||||
id,
|
||||
due: old_due,
|
||||
}));
|
||||
}
|
||||
task.set_status(TaskStatus::at(due as i64));
|
||||
}
|
||||
batch
|
||||
.set(due_key, task_type.to_id().serialize())
|
||||
.set(task_key, task.to_pickled_vec());
|
||||
"Rewrote the queue row from its task."
|
||||
} else {
|
||||
batch.clear(due_key);
|
||||
"Removed a queue row with no task."
|
||||
};
|
||||
|
||||
match server.store().write(batch.build_all()).await {
|
||||
Ok(_) => {
|
||||
repaired = true;
|
||||
trc::event!(
|
||||
TaskManager(TaskManagerEvent::TaskIgnored),
|
||||
Id = id,
|
||||
Due = trc::Value::Timestamp(due),
|
||||
Reason = action,
|
||||
);
|
||||
}
|
||||
Err(err) if err.matches(trc::EventType::Store(trc::StoreEvent::AssertValueFailed)) => {
|
||||
// The task went away meanwhile; the next scan looks again
|
||||
}
|
||||
Err(err) => {
|
||||
trc::error!(
|
||||
err.id(id)
|
||||
.details("Failed to repair an unreadable queue row.")
|
||||
.caused_by(trc::location!())
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
repaired
|
||||
}
|
||||
|
||||
/// inbuxa: the task type a report reschedule meant, when a queue row holds
|
||||
/// an internal report's object type (the value that reschedule wrote).
|
||||
fn rescheduled_report_type(value: &[u8]) -> Option<TaskType> {
|
||||
let id = u16::from_be_bytes(value.get(..2)?.try_into().ok()?);
|
||||
match ObjectType::from_id(id)? {
|
||||
ObjectType::DmarcInternalReport => Some(TaskType::DmarcReport),
|
||||
ObjectType::TlsInternalReport => Some(TaskType::TlsReport),
|
||||
_ => None,
|
||||
}
|
||||
}
|
||||
|
||||
@@ -2,13 +2,15 @@
|
||||
* SPDX-FileCopyrightText: 2020 Stalwart Labs LLC <[email protected]>
|
||||
*
|
||||
* SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL
|
||||
*
|
||||
* Modified by Coffey Labs in 2026 for INBUXA.
|
||||
*/
|
||||
|
||||
use crate::task_manager::{TaskFailureType, TaskResult};
|
||||
use common::{
|
||||
Server,
|
||||
ipc::{BroadcastEvent, RegistryChange},
|
||||
manager::{SPAM_CLASSIFIER_KEY, SPAM_TRAINER_KEY, fetch_resource},
|
||||
manager::{SPAM_CLASSIFIER_KEY, SPAM_TRAINER_KEY, fetch_resource, spam_rules},
|
||||
};
|
||||
use registry::{
|
||||
schema::{
|
||||
@@ -106,6 +108,7 @@ struct RuleUpdateResult {
|
||||
|
||||
async fn update_spam_rules(server: &Server) -> trc::Result<TaskResult> {
|
||||
let started = Instant::now();
|
||||
let bundled = server.core.spam.spam_rules_url.is_none();
|
||||
let rules = match fetch_spam_rules(server).await {
|
||||
Ok(rules) => rules,
|
||||
Err(err) => {
|
||||
@@ -289,29 +292,36 @@ async fn update_spam_rules(server: &Server) -> trc::Result<TaskResult> {
|
||||
Elapsed = started.elapsed(),
|
||||
);
|
||||
|
||||
// inbuxa: so the next start knows these bundled rules are in
|
||||
if bundled {
|
||||
spam_rules::set_applied_version(server.store(), spam_rules::BUNDLED_SPAM_RULES_VERSION)
|
||||
.await?;
|
||||
}
|
||||
|
||||
Ok(TaskResult::Success(vec![]))
|
||||
}
|
||||
|
||||
async fn fetch_spam_rules(server: &Server) -> Result<Rules, RuleUpdateError> {
|
||||
let Some(rules_url) = server.core.spam.spam_rules_url.as_ref() else {
|
||||
return Err(RuleUpdateError {
|
||||
typ: TaskFailureType::Permanent,
|
||||
reason: "Spam rules resource URL not configured".to_string(),
|
||||
});
|
||||
};
|
||||
let rules_json: AHashMap<String, Vec<serde_json::Value>> =
|
||||
fetch_resource(rules_url, None, Duration::from_secs(60), 1024 * 500)
|
||||
// inbuxa: no URL means the rules bundled with the server
|
||||
let bytes = match server.core.spam.spam_rules_url.as_ref() {
|
||||
Some(rules_url) => fetch_resource(rules_url, None, Duration::from_secs(60), 1024 * 500)
|
||||
.await
|
||||
.map_err(|reason| RuleUpdateError {
|
||||
typ: TaskFailureType::Temporary,
|
||||
reason,
|
||||
}),
|
||||
None => spam_rules::bundled_rules().map_err(|reason| RuleUpdateError {
|
||||
typ: TaskFailureType::Permanent,
|
||||
reason,
|
||||
}),
|
||||
};
|
||||
let rules_json: AHashMap<String, Vec<serde_json::Value>> =
|
||||
bytes.and_then(|bytes| {
|
||||
serde_json::from_slice(&bytes).map_err(|err| RuleUpdateError {
|
||||
typ: TaskFailureType::Permanent,
|
||||
reason: format!("Failed to parse spam rules JSON: {err}"),
|
||||
})
|
||||
.and_then(|bytes| {
|
||||
serde_json::from_slice(&bytes).map_err(|err| RuleUpdateError {
|
||||
typ: TaskFailureType::Permanent,
|
||||
reason: format!("Failed to parse spam rules JSON: {err}"),
|
||||
})
|
||||
})?;
|
||||
})?;
|
||||
|
||||
let mut rules = Rules::default();
|
||||
for (object_type, values) in rules_json {
|
||||
|
||||
@@ -2,6 +2,8 @@
|
||||
* SPDX-FileCopyrightText: 2020 Stalwart Labs LLC <[email protected]>
|
||||
*
|
||||
* SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL
|
||||
*
|
||||
* Modified by Coffey Labs in 2026 for INBUXA.
|
||||
*/
|
||||
|
||||
use registry::{
|
||||
@@ -40,35 +42,49 @@ pub trait InternalReportIndex: ObjectImpl {
|
||||
|
||||
fn primary_key(&self) -> ValueClass;
|
||||
|
||||
/// Moves the report's delivery, and its queued task, to `at`.
|
||||
///
|
||||
/// inbuxa: the new queue row carries the task's type, as
|
||||
/// `schedule_task_with_id` writes it, and the task row gets the new due
|
||||
/// too. `queued` is the task as stored: its due, not the report's
|
||||
/// `deliverAt`, is the queue row that exists (they differ once the task
|
||||
/// has been retried).
|
||||
fn reschedule_ops(
|
||||
&mut self,
|
||||
batch: &mut BatchBuilder,
|
||||
item_id: u64,
|
||||
revision: u64,
|
||||
at: UTCDateTime,
|
||||
queued: Option<&Task>,
|
||||
) {
|
||||
let current_deliver_at = self.deliver_at();
|
||||
let current_due = current_deliver_at.timestamp() as u64;
|
||||
let queued_due = queued.map_or(current_due, |task| task.due_timestamp());
|
||||
let new_due = at.timestamp() as u64;
|
||||
|
||||
if current_deliver_at != at {
|
||||
if current_deliver_at != at || queued_due != new_due {
|
||||
let object = Self::OBJECT;
|
||||
let object_id = object.to_id();
|
||||
let key = ValueClass::Registry(RegistryClass::Item { object_id, item_id });
|
||||
|
||||
self.set_deliver_at(at);
|
||||
|
||||
batch
|
||||
.assert_value(key.clone(), AssertValue::Hash(revision))
|
||||
.clear(ValueClass::TaskQueue(TaskQueueClass::Due {
|
||||
batch.assert_value(key.clone(), AssertValue::Hash(revision));
|
||||
if queued_due != new_due {
|
||||
batch.clear(ValueClass::TaskQueue(TaskQueueClass::Due {
|
||||
id: item_id,
|
||||
due: current_deliver_at.timestamp() as u64,
|
||||
}))
|
||||
.set(
|
||||
ValueClass::TaskQueue(TaskQueueClass::Due {
|
||||
id: item_id,
|
||||
due: at.timestamp() as u64,
|
||||
}),
|
||||
object_id.serialize(),
|
||||
)
|
||||
due: queued_due,
|
||||
}));
|
||||
}
|
||||
// A row an earlier reschedule left at the report's deliverAt
|
||||
if current_due != new_due && current_due != queued_due {
|
||||
batch.clear(ValueClass::TaskQueue(TaskQueueClass::Due {
|
||||
id: item_id,
|
||||
due: current_due,
|
||||
}));
|
||||
}
|
||||
batch
|
||||
.schedule_task_with_id(item_id, self.task(item_id))
|
||||
.set(key, self.to_pickled_vec());
|
||||
}
|
||||
}
|
||||
|
||||
@@ -81,7 +81,7 @@ fn legacy_setting(name: &str, is_set: impl Fn(&str) -> bool) -> Option<String> {
|
||||
#[macro_export]
|
||||
macro_rules! brand_version {
|
||||
() => {
|
||||
"2026.9.24"
|
||||
"2026.9.24.4"
|
||||
};
|
||||
}
|
||||
|
||||
|
||||
@@ -102,7 +102,10 @@ Permissions: `sysSpamLlmGet`, `sysSpamLlmUpdate`.
|
||||
- **Tags and scores.** The classifier's tags are ordinary spam tags, scored
|
||||
by `x:SpamTag` entries like every other tag: `Score` (a number), `Discard`
|
||||
or `Reject`. The documented defaults are `LLM_UNSOLICITED_HIGH` 3.0 and
|
||||
`LLM_LEGITIMATE_HIGH` −3.0. A tag with no entry scores 0.
|
||||
`LLM_LEGITIMATE_HIGH` −3.0. A tag with no entry scores 0. The server ships
|
||||
those entries in its bundled spam rules (`resources/spam-filter/`), loaded
|
||||
on first boot and again when the bundled version changes, so an install
|
||||
that predates them gains them on upgrade (added 2026-09-23).
|
||||
- **`interactAi`** permission ("Interact with AI models"): lets an account's
|
||||
own Sieve scripts call `llm_prompt`. This repository's default roles give it
|
||||
to users, tenant administrators and superusers
|
||||
|
||||
Binary file not shown.
@@ -1 +1 @@
|
||||
rWqJwNJkqgKsbC1eqcEmmIAMtJPmnlDlThR2ZtW_N1c
|
||||
VbnFuwCOTBh0s2T-NuRhb2JaJr8Jl5s3LgXv4Pv2sTg
|
||||
@@ -0,0 +1,32 @@
|
||||
# Bundled spam filter rules
|
||||
|
||||
`spam-filter-rules.json.gz` is the published rules file of
|
||||
[spam-filter](https://github.com/stalwartlabs/spam-filter) **v3.0.2**,
|
||||
unmodified. The server embeds it (`crates/common/src/manager/spam_rules.rs`)
|
||||
and loads it whenever no other rules source is configured, so a release
|
||||
scores mail with the rules it was tested with, offline and with nothing to
|
||||
fetch. The rules URL setting stays an operator override.
|
||||
|
||||
The rules are dual-licensed MIT or Apache-2.0, Copyright (C) 2024, Stalwart
|
||||
Labs LLC; the fork takes them under MIT, with the notice in `THIRD-PARTY.md`.
|
||||
|
||||
They include the scores for the AI classifier's tags (`LLM_*`, 3.0 for the
|
||||
high-confidence spam categories, −3.0 for legitimate), which match
|
||||
`docs/spec/features/ai-spam-classification.md`.
|
||||
|
||||
## Updating
|
||||
|
||||
The `upstream-watch` workflow opens an issue when spam-filter publishes a
|
||||
newer release. To take it:
|
||||
|
||||
1. Download `spam-filter-rules.json.gz` from that release, pinned by tag
|
||||
(`releases/download/vX.Y.Z/…`, not `latest`), over this file.
|
||||
2. Set `BUNDLED_SPAM_RULES_VERSION` in `spam_rules.rs` and the version in
|
||||
this README and in `THIRD-PARTY.md`.
|
||||
3. Run the antispam test (`STORE=RocksDb RUST_MIN_STACK=16777216 cargo test
|
||||
-p tests --lib -- smtp::inbound::antispam::antispam --exact`) and fix
|
||||
expectations the new rules change, knowingly.
|
||||
|
||||
On the next start each server loads the new version once. Loading only adds
|
||||
rules and tags that are missing; it never changes an existing one, so an
|
||||
operator's own adjustments survive.
|
||||
@@ -86,19 +86,10 @@ async fn antispam() {
|
||||
.registry_create_object(SpamSettings {
|
||||
score_spam: Float::new(5.0),
|
||||
// inbuxa: the rules carry the scores the expectations are written
|
||||
// against, so they're pinned (spam-filter v3.0.2, beside the test
|
||||
// cases) rather than read from a developer's own checkout, which
|
||||
// left every score at zero. SPAM_RULES_URL still overrides.
|
||||
spam_filter_rules_url: std::env::var("SPAM_RULES_URL")
|
||||
.unwrap_or_else(|_| {
|
||||
concat!(
|
||||
"file://",
|
||||
env!("CARGO_MANIFEST_DIR"),
|
||||
"/resources/smtp/antispam/spam-filter-rules.json.gz"
|
||||
)
|
||||
.to_string()
|
||||
})
|
||||
.into(),
|
||||
// against. Unset, the server uses the rules bundled with it
|
||||
// (resources/spam-filter/), the path production takes;
|
||||
// SPAM_RULES_URL tests another set.
|
||||
spam_filter_rules_url: std::env::var("SPAM_RULES_URL").ok(),
|
||||
..Default::default()
|
||||
})
|
||||
.await;
|
||||
|
||||
@@ -2,9 +2,12 @@
|
||||
* SPDX-FileCopyrightText: 2020 Stalwart Labs LLC <[email protected]>
|
||||
*
|
||||
* SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL
|
||||
*
|
||||
* Modified by Coffey Labs in 2026 for INBUXA.
|
||||
*/
|
||||
|
||||
pub mod analyze;
|
||||
pub mod dmarc;
|
||||
pub mod reschedule; // inbuxa: report reschedules and unreadable queue rows
|
||||
pub mod scheduler;
|
||||
pub mod tls;
|
||||
|
||||
@@ -0,0 +1,370 @@
|
||||
/*
|
||||
* SPDX-FileCopyrightText: 2026 Coffey Labs
|
||||
*
|
||||
* SPDX-License-Identifier: AGPL-3.0-only
|
||||
*/
|
||||
|
||||
//! Rescheduling an internal DMARC or TLS report over JMAP moves its task: the
|
||||
//! task runs at the new time, x:Task/get shows the new due, and tasks due
|
||||
//! after it still run. A task queue row whose type can't be read is logged
|
||||
//! and repaired rather than stopping every task due after it, including the
|
||||
//! rows an earlier reschedule wrote with the report's object type.
|
||||
|
||||
use crate::utils::server::{TestServer, TestServerBuilder};
|
||||
use common::{
|
||||
Server,
|
||||
config::smtp::report::AggregateFrequency,
|
||||
ipc::{DmarcEvent, PolicyType, TlsEvent},
|
||||
};
|
||||
use mail_auth::{
|
||||
common::parse::TxtRecordParser,
|
||||
dmarc::Dmarc,
|
||||
mta_sts::TlsRpt,
|
||||
report::{ActionDisposition, DmarcResult, Record},
|
||||
};
|
||||
use registry::{
|
||||
schema::{
|
||||
enums::{TaskStoreMaintenanceType, TaskType},
|
||||
prelude::{ObjectType, Property},
|
||||
structs::{
|
||||
DmarcInternalReport, DmarcReportSettings, Expression, Task, TaskStatus,
|
||||
TaskStoreMaintenance, TlsInternalReport, TlsReportSettings,
|
||||
},
|
||||
},
|
||||
types::{EnumImpl, ObjectImpl, datetime::UTCDateTime},
|
||||
};
|
||||
use serde_json::json;
|
||||
use smtp::reporting::{index::InternalReportIndex, send::MtaReportSend};
|
||||
use std::{
|
||||
sync::Arc,
|
||||
time::{Duration, Instant},
|
||||
};
|
||||
use store::{
|
||||
SerializeInfallible, ValueKey,
|
||||
write::{BatchBuilder, RegistryClass, TaskQueueClass, ValueClass, now},
|
||||
};
|
||||
use types::id::Id;
|
||||
use utils::snowflake::SnowflakeIdGenerator;
|
||||
|
||||
#[tokio::test(flavor = "multi_thread")]
|
||||
#[serial_test::serial]
|
||||
async fn report_reschedule() {
|
||||
let mut test = TestServerBuilder::new("smtp_report_reschedule")
|
||||
.await
|
||||
.with_http_listener(19057)
|
||||
.await
|
||||
.capture_queue()
|
||||
.build()
|
||||
.await;
|
||||
|
||||
let admin = test.account("admin");
|
||||
admin
|
||||
.registry_create_object(TlsReportSettings {
|
||||
max_report_size: Expression {
|
||||
else_: "1024".into(),
|
||||
..Default::default()
|
||||
},
|
||||
..Default::default()
|
||||
})
|
||||
.await;
|
||||
admin
|
||||
.registry_create_object(DmarcReportSettings {
|
||||
aggregate_max_report_size: Expression {
|
||||
else_: "1024".into(),
|
||||
..Default::default()
|
||||
},
|
||||
..Default::default()
|
||||
})
|
||||
.await;
|
||||
admin.reload_settings().await;
|
||||
test.reload_core();
|
||||
test.expect_reload_settings().await;
|
||||
let admin = test.account("admin");
|
||||
|
||||
// A daily DMARC and TLS report, due a day from now
|
||||
schedule_dmarc(&test, "foobar.org").await;
|
||||
schedule_tls(&test, "foobar.org").await;
|
||||
let dmarc_id = wait_for_report::<DmarcInternalReport>(&test, "foobar.org").await;
|
||||
let tls_id = wait_for_report::<TlsInternalReport>(&test, "foobar.org").await;
|
||||
|
||||
// Reschedule both to a few seconds from now, with a task due after them
|
||||
let at = now() + 3;
|
||||
let later = marker_task(&test.server, at + 3).await;
|
||||
for (object, id, task_type) in [
|
||||
(
|
||||
ObjectType::DmarcInternalReport,
|
||||
dmarc_id,
|
||||
TaskType::DmarcReport,
|
||||
),
|
||||
(ObjectType::TlsInternalReport, tls_id, TaskType::TlsReport),
|
||||
] {
|
||||
admin
|
||||
.registry_update_object(
|
||||
object,
|
||||
id,
|
||||
json!({
|
||||
Property::DeliverAt: UTCDateTime::from_timestamp(at as i64),
|
||||
}),
|
||||
)
|
||||
.await;
|
||||
|
||||
// x:Task/get shows the new due, and the queue row carries the task's
|
||||
// type. Upstream wrote the report's object type there and left the
|
||||
// task at its old due
|
||||
let task = admin.registry_get::<Task>(id).await;
|
||||
assert_eq!(task.object_type(), task_type);
|
||||
assert_eq!(
|
||||
task.due_timestamp(),
|
||||
at,
|
||||
"{object:?} task due not moved: {task:?}"
|
||||
);
|
||||
assert_eq!(
|
||||
queue_row(&test.server, id.id(), at).await,
|
||||
Some(task_type.to_id().serialize()),
|
||||
"{object:?} queue row"
|
||||
);
|
||||
}
|
||||
|
||||
// Both reports go out at the new time, and the later task still runs
|
||||
wait_until_run(&test.server, &[dmarc_id.id(), tls_id.id(), later]).await;
|
||||
assert!(now() >= at, "the reports went out before their new time");
|
||||
assert!(
|
||||
admin
|
||||
.registry_get_all::<DmarcInternalReport>()
|
||||
.await
|
||||
.is_empty()
|
||||
);
|
||||
assert!(
|
||||
admin
|
||||
.registry_get_all::<TlsInternalReport>()
|
||||
.await
|
||||
.is_empty()
|
||||
);
|
||||
|
||||
// Rows an earlier reschedule may have left in a store: one with the
|
||||
// report's object type and the task left at its old due, and one that
|
||||
// is unreadable and has no task behind it. Neither may hold back a task
|
||||
// due after them.
|
||||
schedule_dmarc(&test, "foobar.net").await;
|
||||
let dmarc_id = wait_for_report::<DmarcInternalReport>(&test, "foobar.net").await;
|
||||
let at = now() + 2;
|
||||
let old_due = old_style_reschedule(&test.server, dmarc_id.id(), at).await;
|
||||
let orphan = SnowflakeIdGenerator::global_id().unwrap();
|
||||
let mut batch = BatchBuilder::new();
|
||||
batch.set(
|
||||
ValueClass::TaskQueue(TaskQueueClass::Due {
|
||||
id: orphan,
|
||||
due: at,
|
||||
}),
|
||||
vec![0xff, 0xff],
|
||||
);
|
||||
test.server.store().write(batch.build_all()).await.unwrap();
|
||||
let later = marker_task(&test.server, at + 2).await;
|
||||
|
||||
wait_until_run(&test.server, &[dmarc_id.id(), later]).await;
|
||||
assert!(
|
||||
admin
|
||||
.registry_get_all::<DmarcInternalReport>()
|
||||
.await
|
||||
.is_empty()
|
||||
);
|
||||
assert_eq!(queue_row(&test.server, orphan, at).await, None);
|
||||
assert_eq!(queue_row(&test.server, dmarc_id.id(), at).await, None);
|
||||
assert_eq!(queue_row(&test.server, dmarc_id.id(), old_due).await, None);
|
||||
|
||||
// x:Task/query by type skips an unreadable row rather than failing
|
||||
let mut batch = BatchBuilder::new();
|
||||
let due = now() + 3600;
|
||||
batch.set(
|
||||
ValueClass::TaskQueue(TaskQueueClass::Due { id: orphan, due }),
|
||||
vec![0xff, 0xff],
|
||||
);
|
||||
test.server.store().write(batch.build_all()).await.unwrap();
|
||||
admin
|
||||
.registry_query_ids(
|
||||
ObjectType::Task,
|
||||
vec![(Property::Type, TaskType::DmarcReport.as_str())],
|
||||
Vec::<&str>::new(),
|
||||
)
|
||||
.await;
|
||||
let mut batch = BatchBuilder::new();
|
||||
batch.clear(ValueClass::TaskQueue(TaskQueueClass::Due {
|
||||
id: orphan,
|
||||
due,
|
||||
}));
|
||||
test.server.store().write(batch.build_all()).await.unwrap();
|
||||
|
||||
if test.is_reset() {
|
||||
test.temp_dir.delete();
|
||||
}
|
||||
}
|
||||
|
||||
async fn schedule_dmarc(test: &TestServer, domain: &str) {
|
||||
test.server
|
||||
.schedule_report(DmarcEvent {
|
||||
domain: domain.to_string(),
|
||||
report_record: Record::new()
|
||||
.with_source_ip("192.168.1.2".parse().unwrap())
|
||||
.with_action_disposition(ActionDisposition::Pass)
|
||||
.with_dmarc_dkim_result(DmarcResult::Pass)
|
||||
.with_dmarc_spf_result(DmarcResult::Fail)
|
||||
.with_envelope_from("[email protected]")
|
||||
.with_envelope_to("[email protected]")
|
||||
.with_header_from("[email protected]"),
|
||||
dmarc_record: Arc::new(
|
||||
Dmarc::parse(format!("v=DMARC1; p=reject; rua=mailto:reports@{domain}").as_bytes())
|
||||
.unwrap(),
|
||||
),
|
||||
interval: AggregateFrequency::Daily,
|
||||
span_id: 0,
|
||||
})
|
||||
.await;
|
||||
}
|
||||
|
||||
async fn schedule_tls(test: &TestServer, domain: &str) {
|
||||
test.server
|
||||
.schedule_report(TlsEvent {
|
||||
domain: domain.to_string(),
|
||||
policy: PolicyType::None,
|
||||
failure: None,
|
||||
tls_record: Arc::new(
|
||||
TlsRpt::parse(format!("v=TLSRPTv1;rua=mailto:reports@{domain}").as_bytes())
|
||||
.unwrap(),
|
||||
),
|
||||
interval: AggregateFrequency::Daily,
|
||||
span_id: 0,
|
||||
})
|
||||
.await;
|
||||
}
|
||||
|
||||
trait ReportDomain: ObjectImpl {
|
||||
fn report_domain(&self) -> &str;
|
||||
}
|
||||
|
||||
impl ReportDomain for DmarcInternalReport {
|
||||
fn report_domain(&self) -> &str {
|
||||
&self.domain
|
||||
}
|
||||
}
|
||||
|
||||
impl ReportDomain for TlsInternalReport {
|
||||
fn report_domain(&self) -> &str {
|
||||
&self.domain
|
||||
}
|
||||
}
|
||||
|
||||
async fn wait_for_report<T: ReportDomain>(test: &TestServer, domain: &str) -> Id {
|
||||
let admin = test.account("admin");
|
||||
for _ in 0..100 {
|
||||
if let Some((id, _)) = admin
|
||||
.registry_get_all::<T>()
|
||||
.await
|
||||
.into_iter()
|
||||
.find(|(_, report)| report.report_domain() == domain)
|
||||
{
|
||||
return id;
|
||||
}
|
||||
tokio::time::sleep(Duration::from_millis(100)).await;
|
||||
}
|
||||
panic!("No {} for {domain}", T::OBJECT.as_str());
|
||||
}
|
||||
|
||||
/// A task that succeeds when it runs, due at `due`.
|
||||
async fn marker_task(server: &Server, due: u64) -> u64 {
|
||||
let id = SnowflakeIdGenerator::global_id().unwrap();
|
||||
let mut batch = BatchBuilder::new();
|
||||
batch.schedule_task_with_id(
|
||||
id,
|
||||
Task::StoreMaintenance(TaskStoreMaintenance {
|
||||
maintenance_type: TaskStoreMaintenanceType::RemoveLockDav,
|
||||
shard_index: Some(0),
|
||||
status: TaskStatus::at(due as i64),
|
||||
}),
|
||||
);
|
||||
server.store().write(batch.build_all()).await.unwrap();
|
||||
server.notify_task_queue();
|
||||
id
|
||||
}
|
||||
|
||||
/// What the reschedule before this fix wrote: the report's object type in
|
||||
/// the new queue row, and the task row left at its old due. Returns that
|
||||
/// old due.
|
||||
async fn old_style_reschedule(server: &Server, item_id: u64, at: u64) -> u64 {
|
||||
let object_id = ObjectType::DmarcInternalReport.to_id();
|
||||
let key = ValueClass::Registry(RegistryClass::Item { object_id, item_id });
|
||||
let mut report = server
|
||||
.store()
|
||||
.get_value::<DmarcInternalReport>(ValueKey::from(key.clone()))
|
||||
.await
|
||||
.unwrap()
|
||||
.unwrap();
|
||||
let old_due = report.deliver_at().timestamp() as u64;
|
||||
report.set_deliver_at(UTCDateTime::from_timestamp(at as i64));
|
||||
let mut batch = BatchBuilder::new();
|
||||
batch
|
||||
.clear(ValueClass::TaskQueue(TaskQueueClass::Due {
|
||||
id: item_id,
|
||||
due: old_due,
|
||||
}))
|
||||
.set(
|
||||
ValueClass::TaskQueue(TaskQueueClass::Due {
|
||||
id: item_id,
|
||||
due: at,
|
||||
}),
|
||||
object_id.serialize(),
|
||||
)
|
||||
.set(key, report.to_pickled_vec());
|
||||
server.store().write(batch.build_all()).await.unwrap();
|
||||
server.notify_task_queue();
|
||||
old_due
|
||||
}
|
||||
|
||||
struct RawValue(Vec<u8>);
|
||||
|
||||
impl store::Deserialize for RawValue {
|
||||
fn deserialize(bytes: &[u8]) -> trc::Result<Self> {
|
||||
Ok(RawValue(bytes.to_vec()))
|
||||
}
|
||||
}
|
||||
|
||||
async fn queue_row(server: &Server, id: u64, due: u64) -> Option<Vec<u8>> {
|
||||
server
|
||||
.store()
|
||||
.get_value::<RawValue>(ValueKey::from(ValueClass::TaskQueue(TaskQueueClass::Due {
|
||||
id,
|
||||
due,
|
||||
})))
|
||||
.await
|
||||
.unwrap()
|
||||
.map(|raw| raw.0)
|
||||
}
|
||||
|
||||
async fn task_exists(server: &Server, id: u64) -> bool {
|
||||
server
|
||||
.store()
|
||||
.get_value::<Task>(ValueKey::from(ValueClass::TaskQueue(
|
||||
TaskQueueClass::Task { id },
|
||||
)))
|
||||
.await
|
||||
.unwrap()
|
||||
.is_some()
|
||||
}
|
||||
|
||||
async fn wait_until_run(server: &Server, ids: &[u64]) {
|
||||
let started = Instant::now();
|
||||
loop {
|
||||
let mut pending = Vec::new();
|
||||
for id in ids {
|
||||
if task_exists(server, *id).await {
|
||||
pending.push(*id);
|
||||
}
|
||||
}
|
||||
if pending.is_empty() {
|
||||
return;
|
||||
}
|
||||
if started.elapsed() > Duration::from_secs(30) {
|
||||
panic!("tasks {pending:?} never ran");
|
||||
}
|
||||
tokio::time::sleep(Duration::from_millis(200)).await;
|
||||
}
|
||||
}
|
||||
Executable
+120
@@ -0,0 +1,120 @@
|
||||
#!/usr/bin/env python3
|
||||
# SPDX-FileCopyrightText: 2026 Coffey Labs
|
||||
# SPDX-License-Identifier: AGPL-3.0-or-later
|
||||
"""Every path Cargo patches has to be in the image's build context.
|
||||
|
||||
Cargo.toml's [patch.crates-io] can point at a directory in this repository,
|
||||
and the Dockerfile builds from a context that .dockerignore prunes to almost
|
||||
nothing. Those two facts met on 2026-09-23: a vendored, patched sieve-rs
|
||||
landed, CI stayed green -- it builds from a checkout, where the directory is
|
||||
simply there -- and the release build failed on
|
||||
|
||||
failed to load source for dependency `sieve-rs`
|
||||
failed to read /build/vendor/sieve-rs/Cargo.toml
|
||||
|
||||
after a tag had already been pushed. This is seconds, and it runs beside the
|
||||
other fork checks rather than waiting for a release to find out.
|
||||
|
||||
Being in the context isn't enough on its own: the Dockerfile cooks the
|
||||
dependencies (`cargo chef cook`) before it copies the tree in, from a recipe
|
||||
that carries only the workspace's manifests. So each patched path must also be
|
||||
copied into that stage before the cook step, or the same error comes back
|
||||
there -- as it did for 2026.9.24.2, the first tag after the context fix.
|
||||
"""
|
||||
|
||||
import re
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
root = Path(__file__).resolve().parents[2]
|
||||
|
||||
|
||||
def patched_paths(manifest: Path) -> list[str]:
|
||||
"""Directories named by a [patch...] section's `path = "..."` entries."""
|
||||
out, in_patch = [], False
|
||||
for line in manifest.read_text().splitlines():
|
||||
stripped = line.strip()
|
||||
if stripped.startswith("["):
|
||||
in_patch = stripped.startswith("[patch")
|
||||
continue
|
||||
if not in_patch:
|
||||
continue
|
||||
m = re.search(r'path\s*=\s*"([^"]+)"', stripped)
|
||||
if m:
|
||||
out.append(m.group(1))
|
||||
return out
|
||||
|
||||
|
||||
def allowed(dockerignore: Path) -> set[str]:
|
||||
"""The first path segment of every re-inclusion rule."""
|
||||
keep = set()
|
||||
for line in dockerignore.read_text().splitlines():
|
||||
stripped = line.strip()
|
||||
if stripped.startswith("!"):
|
||||
keep.add(stripped[1:].strip("/").split("/")[0])
|
||||
return keep
|
||||
|
||||
|
||||
def copied_before_cook(dockerfile: Path) -> list[str] | None:
|
||||
"""Sources COPY'd into the stage that runs `cargo chef cook`, before it.
|
||||
|
||||
None when no stage cooks. A `COPY . .` covers everything.
|
||||
"""
|
||||
stage: list[str] = []
|
||||
for line in dockerfile.read_text().splitlines():
|
||||
stripped = line.strip()
|
||||
if re.match(r"(?i)^FROM\s", stripped):
|
||||
stage = []
|
||||
continue
|
||||
if "cargo chef cook" in stripped:
|
||||
return stage
|
||||
m = re.match(r"(?i)^COPY\s+(?!--from)(.+)$", stripped)
|
||||
if m:
|
||||
parts = m.group(1).split()
|
||||
stage.extend(p.strip("./").split("/")[0] or "." for p in parts[:-1])
|
||||
return None
|
||||
|
||||
|
||||
def main() -> int:
|
||||
paths = patched_paths(root / "Cargo.toml")
|
||||
if not paths:
|
||||
print("no patched paths to check")
|
||||
return 0
|
||||
keep = allowed(root / ".dockerignore")
|
||||
bad = []
|
||||
for p in paths:
|
||||
top = p.strip("/").split("/")[0]
|
||||
if top not in keep:
|
||||
bad.append((p, top))
|
||||
elif not (root / p).is_dir():
|
||||
bad.append((p, None))
|
||||
for path, top in bad:
|
||||
if top is None:
|
||||
print(f"Cargo.toml patches {path}, which does not exist", file=sys.stderr)
|
||||
else:
|
||||
print(
|
||||
f"Cargo.toml patches {path}, but .dockerignore does not re-include {top!r}:\n"
|
||||
f" the image build would not see it, and cargo would fail on it.\n"
|
||||
f" Add `!{top}` to .dockerignore.",
|
||||
file=sys.stderr,
|
||||
)
|
||||
copied = copied_before_cook(root / "Dockerfile")
|
||||
if copied is not None and "." not in copied:
|
||||
for p in paths:
|
||||
top = p.strip("/").split("/")[0]
|
||||
if top not in copied:
|
||||
print(
|
||||
f"Cargo.toml patches {p}, but the Dockerfile doesn't copy {top!r} into the\n"
|
||||
f" stage that runs `cargo chef cook` before that step, so cooking the\n"
|
||||
f" dependencies fails on it. Add `COPY {top}/ {top}/` before the cook.",
|
||||
file=sys.stderr,
|
||||
)
|
||||
bad.append((p, top))
|
||||
if bad:
|
||||
return 1
|
||||
print(f"build context and cook stage include every patched path: {', '.join(paths)}")
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
@@ -23,6 +23,6 @@ crates/migration/src/lib.rs "STALWART_SPAM_CLASSIFIER_MODEL.lz4"
|
||||
crates/migration/src/lib.rs "STALWART_SPAM_TRAIN_DATA.lz4"
|
||||
crates/types/src/branding.rs "STALWART"
|
||||
|
||||
# OPEN, not yet decided (2026-09-22): upstream's published spam-filter rules,
|
||||
# which the server downloads at runtime from this address.
|
||||
crates/registry/src/schema/structs_impl.rs "https://github.com/stalwartlabs/spam-filter/releases/latest/download/spam-filter-rules.json.gz"
|
||||
# Upstream's old default rules source, read only to treat it as unset: the
|
||||
# server uses the rules bundled with it (resources/spam-filter/).
|
||||
crates/common/src/manager/spam_rules.rs "https://github.com/stalwartlabs/spam-filter/releases/latest/download/spam-filter-rules.json.gz"
|
||||
|
||||
@@ -46,6 +46,10 @@ TEXT_RENAMES = [
|
||||
# that must match their containers and identity provider (database users,
|
||||
# passwords, an OIDC audience), and name their databases explicitly.
|
||||
('"stalwart".to_string()', '"inbuxa".to_string()', ('crates',)),
|
||||
# The spam filter rules ship with the server (common::manager::spam_rules);
|
||||
# upstream's default of fetching its latest from GitHub becomes unset.
|
||||
('spam_filter_rules_url: Some("https://github.com/stalwartlabs/spam-filter/releases/latest/download/spam-filter-rules.json.gz".to_string()),',
|
||||
'spam_filter_rules_url: None,', ('crates',)),
|
||||
]
|
||||
ROOTS = ('crates', 'tests', 'resources')
|
||||
SKIP_SUFFIXES = {'.md', '.txt'}
|
||||
@@ -58,6 +62,10 @@ SCHEMA_HASH = Path('resources/schema/schema.json.sha256')
|
||||
SCHEMA_RENAMES = [
|
||||
('"stalwart"', '"inbuxa"'),
|
||||
('vnd.stalwart', 'vnd.inbuxa'),
|
||||
# The bundled spam rules: no default URL, and say what empty means.
|
||||
('"spamFilterRulesUrl":"https://github.com/stalwartlabs/spam-filter/releases/latest/download/spam-filter-rules.json.gz",', ''),
|
||||
('"URL to download spam filter rules from"',
|
||||
'"URL to download spam filter rules from. Empty uses the rules bundled with the server."'),
|
||||
]
|
||||
|
||||
|
||||
|
||||
Reference in New Issue
Block a user