fix(stalwart): harden the platform admin forward after the skeptic round

The users cache no longer demands a mailbox on every host (a workstation
inventory without mail broke every users lookup) and skips forwards that
are still unrendered Jinja; a non-mapping target names itself. Stalwart now
enforces the mailbox where it consumes the forward. Under SSO 16_forwards
skips ReloadSettings, which would open the OIDC issuer before Keycloak runs;
the OIDC restart applies the limits. The routing probe treats transient
IMAP/SMTP errors as not yet within the same budget and only checks forwards
in Stalwart's own rounds, so the 67 roles that deploy it as mail provider
keep skipping it.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
kevinveenbirkenbachandClaude Opus 5.5 committed 2026-10-31 23:55:09 +01:00
1 parent bbd8ad2764
commit c1cd4325d6
10 files changed
+282 -44

No files matched your search

+63 -12
View File
@@ -216,11 +216,9 @@ class TestBuildUsers(unittest.TestCase):
self.assertEqual(users["plain"]["forward"], "")
self.assertEqual(users["pinned"]["forward"], "adminuser")
def test_forward_to_a_user_without_mailbox_raises(self):
defs = OrderedDict(
[("plain", {}), ("pinned", {"accounts": [], "forward": "plain"})]
)
with self.assertRaisesRegex(ValueError, "'plain', which carries no mailbox"):
def test_forward_to_an_unknown_user_raises(self):
defs = OrderedDict([("pinned", {"accounts": [], "forward": "nobody"})])
with self.assertRaisesRegex(ValueError, "'pinned' forwards to unknown user"):
cache_users._build_users(
defs, primary_domain="x.example", start_id=1001, become_pwd="pw"
)
@@ -234,25 +232,41 @@ class TestValidateForwards(unittest.TestCase):
def test_users_without_forward_pass(self):
cache_users._validate_forwards({"a": self._user(), "b": self._user()})
def test_forward_to_a_mailbox_passes(self):
def test_forward_to_a_known_user_passes(self):
cache_users._validate_forwards(
{"a": self._user("b", accounts=()), "b": self._user()}
)
def test_chain_of_forwards_to_mailboxes_passes(self):
def test_chain_of_forwards_passes(self):
cache_users._validate_forwards(
{"a": self._user("b", accounts=()), "b": self._user("c"), "c": self._user()}
)
def test_forward_to_a_user_without_mailbox_passes(self):
cache_users._validate_forwards(
{"a": self._user("b"), "b": self._user(accounts=("host", "identity"))}
)
def test_forward_to_unknown_user_raises(self):
with self.assertRaisesRegex(ValueError, "'a' forwards to unknown user 'x'"):
cache_users._validate_forwards({"a": self._user("x")})
def test_forward_to_user_without_mailbox_raises(self):
with self.assertRaisesRegex(ValueError, "carries no mailbox account"):
cache_users._validate_forwards(
{"a": self._user("b"), "b": self._user(accounts=("identity",))}
)
def test_unrendered_forward_is_left_to_the_consumer(self):
cache_users._validate_forwards({"a": self._user("{{ 'nobody' }}")})
def test_unrendered_statement_forward_is_left_to_the_consumer(self):
cache_users._validate_forwards({"a": self._user("{% if x %}nobody{% endif %}")})
def test_unrendered_forward_ends_the_chain(self):
cache_users._validate_forwards(
{"a": self._user("b"), "b": self._user("{{ 'a' }}")}
)
def test_forward_to_a_non_mapping_entry_raises_a_clear_error(self):
with self.assertRaisesRegex(
TypeError, "'a' forwards to 'b', whose entry is None"
):
cache_users._validate_forwards({"a": self._user("b"), "b": None})
def test_forward_to_itself_raises(self):
with self.assertRaisesRegex(ValueError, "Forward loop: a -> a"):
@@ -454,6 +468,43 @@ class TestGetMergedUsersWithOverrides(unittest.TestCase):
)
self.assertEqual(merged["foo"]["forward"], "inbox")
def test_runtime_target_without_mailbox_keeps_the_lookup(self):
with tempfile.TemporaryDirectory() as tmp:
roles = _seed_minimal_user_role(Path(tmp), "web-app-foo")
merged = cache_users.get_merged_users(
variables={
"users": {
"foo": {"forward": "workstation"},
"workstation": {"accounts": ["host", "identity"]},
}
},
roles_dir=roles,
templar=None,
)
self.assertEqual(merged["workstation"]["accounts"], ["host", "identity"])
def test_runtime_templated_forward_is_not_rejected(self):
with tempfile.TemporaryDirectory() as tmp:
roles = _seed_minimal_user_role(Path(tmp), "web-app-foo")
merged = cache_users.get_merged_users(
variables={"users": {"foo": {"forward": "{{ 'inbox' }}"}}},
roles_dir=roles,
templar=None,
)
self.assertEqual(merged["foo"]["forward"], "{{ 'inbox' }}")
def test_runtime_null_target_raises_a_clear_error(self):
with tempfile.TemporaryDirectory() as tmp:
roles = _seed_minimal_user_role(Path(tmp), "web-app-foo")
with self.assertRaisesRegex(
TypeError, "'foo' forwards to 'inbox', whose entry"
):
cache_users.get_merged_users(
variables={"users": {"foo": {"forward": "inbox"}, "inbox": None}},
roles_dir=roles,
templar=None,
)
class TestGetMergedUsersTokenSource(unittest.TestCase):
def setUp(self) -> None: