kevinveenbirkenbach/computer-playbook
1
0 Fork 0
mirror of https://github.com/kevinveenbirkenbach/computer-playbook.git synced 2026-11-06 19:01:54 +00:00
Code Issues Packages Projects Releases Wiki Activity
Files
8a29ba0bfdfafaabdfbc839c557085d86ee24527
computer-playbook/tests/unit/python/plugins/lookup/test_email.py
T

601 lines
23 KiB
Python
Raw Blame History

# nocheck: comments-valid explanatory WHY-comments predate the stricter lint
from __future__ import annotations
import os
import tempfile
import unittest
from pathlib import Path
from typing import ClassVar
from unittest.mock import patch
from ansible.errors import AnsibleError
from plugins.lookup.email import LookupModule
from utils.cache import _reset_cache_for_tests
from utils.cache import base as cache_base
from utils.cache import users as cache_users
from utils.cache.yaml import dump_yaml_str
from utils.roles.mapping import ROLE_FILE_META_DOMAINS, ROLE_FILE_META_SERVICES
def _write_role_config(base_dir: Path, role_name: str, payload: dict) -> None:
config_path = base_dir / "roles" / role_name / ROLE_FILE_META_SERVICES
config_path.parent.mkdir(parents=True, exist_ok=True)
config_path.write_text(dump_yaml_str(payload), encoding="utf-8")
class TestEmailLookup(unittest.TestCase):
@classmethod
def setUpClass(cls) -> None:
_reset_cache_for_tests()
cls.addClassCleanup(_reset_cache_for_tests)
def setUp(self) -> None:
cache_base._reset()
self.lookup = LookupModule()
self.lookup._templar = None
self._cwd = str(Path.cwd())
self._tmpdir = tempfile.TemporaryDirectory()
self._tmp = Path(self._tmpdir.name)
(self._tmp / "roles").mkdir(parents=True, exist_ok=True)
os.chdir(self._tmp)
self._tokens_store_patcher = patch.object(
cache_users, "_load_store_users", return_value={}
)
self._tokens_store_patcher.start()
def tearDown(self) -> None:
self._tokens_store_patcher.stop()
os.chdir(self._cwd)
self._tmpdir.cleanup()
def test_returns_plugin_defaults_when_no_vars(self) -> None:
result = self.lookup.run([], variables={"inventory_hostname": "host1"})[0]
self.assertTrue(result["enabled"])
self.assertEqual(result["host"], "localhost")
self.assertEqual(result["port"], 25)
self.assertEqual(result["from"], "root@host1.localdomain")
self.assertEqual(result["username"], "root@host1.localdomain")
self.assertEqual(result["password"], "")
self.assertFalse(result["tls"])
self.assertFalse(result["auth"])
self.assertTrue(result["smtp"])
def test_host_is_localhost_when_not_external_under_tls(self) -> None:
result = self.lookup.run(
[],
variables={"inventory_hostname": "host1", "TLS_ENABLED": True},
)[0]
self.assertEqual(result["host"], "localhost")
self.assertFalse(result["auth"])
self.assertFalse(result["tls"])
def test_keys_are_lowercased_without_prefix(self) -> None:
result = self.lookup.run([], variables={})[0]
for key in result:
self.assertFalse(key.startswith("SYSTEM_EMAIL_"))
self.assertEqual(key, key.lower())
def test_group_var_overrides_plugin_default(self) -> None:
variables = {
"SYSTEM_EMAIL_HOST": "smtp.example.org",
"SYSTEM_EMAIL_PORT": 465,
"SYSTEM_EMAIL_TLS": True,
"inventory_hostname": "host1",
}
result = self.lookup.run([], variables=variables)[0]
self.assertEqual(result["host"], "smtp.example.org")
self.assertEqual(result["port"], 465)
self.assertTrue(result["tls"])
self.assertEqual(result["from"], "root@host1.localdomain")
def test_onion_domain_disables_tls_but_keeps_auth(self) -> None:
"""An onion mail host has no TLS certificate, so the relay speaks
plaintext — Tor already encrypts the transport. AUTH must survive that:
without credentials the message reaches Mailu unauthenticated with a
local envelope sender, and rspamd's antispoof rule rejects it with
554 5.7.1. Plaintext submission is port 587; 25 is the inbound MX and
offers no AUTH at all."""
variables = {
"MAIL_PROVIDER": "web-app-mailu",
"groups": {"web-app-mailu": ["host1"]},
"SYSTEM_EMAIL_HOST": "mail.abc123.onion",
"TLS_ENABLED": True,
"DOMAIN_PRIMARY": "abc123.onion",
"inventory_hostname": "host1",
}
result = self.lookup.run([], variables=variables)[0]
self.assertFalse(result["tls"])
self.assertTrue(result["auth"])
self.assertEqual(result["port"], 587)
def test_clearnet_external_keeps_tls(self) -> None:
variables = {
"MAIL_PROVIDER": "web-app-mailu",
"groups": {"web-app-mailu": ["host1"]},
"SYSTEM_EMAIL_HOST": "mail.example.org",
"TLS_ENABLED": True,
"DOMAIN_PRIMARY": "example.org",
"inventory_hostname": "host1",
}
result = self.lookup.run([], variables=variables)[0]
self.assertTrue(result["tls"])
self.assertEqual(result["port"], 465)
def test_the_axis_matrix_stays_where_it_was_measured(self) -> None:
"""Decoupling auth from tls may only move the two cases it was meant to.
The clearnet default and the local relay must render byte-identically,
or a change aimed at the onion has silently reached every deploy."""
cases = [
("mail.example.org", True, True, {"tls": True, "auth": True, "port": 465}),
(
"mail.example.org",
False,
True,
{"tls": False, "auth": True, "port": 587},
),
("mail.x.onion", True, True, {"tls": False, "auth": True, "port": 587}),
(
"mail.example.org",
True,
False,
{"tls": False, "auth": False, "port": 25},
),
]
for host, tls_enabled, external, expected in cases:
with self.subTest(host=host, tls=tls_enabled, external=external):
variables = {
"MAIL_PROVIDER": "web-app-mailu",
"groups": {"web-app-mailu": ["host1"]} if external else {},
"SYSTEM_EMAIL_HOST": host,
"TLS_ENABLED": tls_enabled,
"DOMAIN_PRIMARY": host.split(".", 1)[1],
"inventory_hostname": "host1",
}
result = self.lookup.run([], variables=variables)[0]
for key, value in expected.items():
self.assertEqual(
bool(result[key]) if key != "port" else result[key], value
)
def test_auth_mechanism_is_named_only_where_msmtp_would_refuse(self) -> None:
"""msmtp aborts with EX_UNAVAILABLE when it has to pick a mechanism
itself on a cleartext link, so the onion relay needs PLAIN spelled out.
The TLS and the no-relay cases must keep rendering on and off, or the
change reaches deploys it was never meant to touch."""
cases = [
("mail.example.org", True, True, "on"),
("mail.x.onion", True, True, "plain"),
("mail.example.org", True, False, "off"),
]
for host, tls_enabled, external, expected in cases:
with self.subTest(host=host, external=external):
variables = {
"MAIL_PROVIDER": "web-app-mailu",
"groups": {"web-app-mailu": ["host1"]} if external else {},
"SYSTEM_EMAIL_HOST": host,
"TLS_ENABLED": tls_enabled,
"DOMAIN_PRIMARY": host.split(".", 1)[1],
"inventory_hostname": "host1",
}
result = self.lookup.run([], variables=variables)[0]
self.assertEqual(result["auth_mechanism"], expected)
def test_empty_string_falls_back_to_plugin_default(self) -> None:
variables = {
"SYSTEM_EMAIL_HOST": "",
"inventory_hostname": "host1",
}
result = self.lookup.run([], variables=variables)[0]
self.assertEqual(result["host"], "localhost")
def test_application_override_wins_over_defaults(self) -> None:
# Per the file root of meta/services.yml IS the services map
# (no `compose.services` envelope).
_write_role_config(
self._tmp,
"web-app-x",
{"email": {"host": "smtp.app.org", "port": 587}},
)
variables = {
"SYSTEM_EMAIL_HOST": "smtp.global.org",
"SYSTEM_EMAIL_PORT": 25,
"inventory_hostname": "host1",
}
result = self.lookup.run(
["web-app-x"],
variables=variables,
roles_dir=str(self._tmp / "roles"),
)[0]
self.assertEqual(result["host"], "smtp.app.org")
self.assertEqual(result["port"], 587)
self.assertEqual(result["from"], "root@host1.localdomain")
def test_missing_application_returns_defaults(self) -> None:
variables = {
"SYSTEM_EMAIL_HOST": "smtp.global.org",
"inventory_hostname": "host1",
}
result = self.lookup.run(
["web-app-unknown"],
variables=variables,
roles_dir=str(self._tmp / "roles"),
)[0]
self.assertEqual(result["host"], "smtp.global.org")
def test_application_without_email_service_returns_defaults(self) -> None:
_write_role_config(
self._tmp,
"web-app-nomail",
{"logout": {"enabled": True}},
)
variables = {
"SYSTEM_EMAIL_HOST": "smtp.global.org",
"inventory_hostname": "host1",
}
result = self.lookup.run(
["web-app-nomail"],
variables=variables,
roles_dir=str(self._tmp / "roles"),
)[0]
self.assertEqual(result["host"], "smtp.global.org")
def test_too_many_terms_raises(self) -> None:
with self.assertRaises(AnsibleError):
self.lookup.run(["a", "b"], variables={})
def test_an_onion_relay_never_announces_starttls(self) -> None:
# No CA issues a certificate for a .onion name, so a client that
# verifies the peer can never complete the handshake -- Ruby's
# Net::SMTP drops the mail after three silent retries. Tor already
# carries the confidentiality, which is why `tls` is False here too.
_write_role_config(
self._tmp,
"web-app-mailprov",
{"sso": {"oidc": {"submission_via_relay": True}}},
)
variables = {
"MAIL_PROVIDER": "web-app-mailprov",
"group_names": ["web-app-mailprov", "web-app-keycloak"],
"groups": {"web-app-mailprov": ["host1"], "web-app-keycloak": ["host1"]},
"TLS_ENABLED": True,
"SYSTEM_EMAIL_HOST": "mail." + "a" * 56 + ".onion",
"inventory_hostname": "host1",
}
result = self.lookup.run(
[], variables=variables, roles_dir=str(self._tmp / "roles")
)[0]
self.assertFalse(result["start_tls"])
self.assertFalse(result["tls"])
def test_sso_relay_provider_disables_auth_and_uses_port_25(self) -> None:
# submission_via_relay + Keycloak deployed -> relay on 25, no auth, STARTTLS.
_write_role_config(
self._tmp,
"web-app-mailprov",
{"sso": {"oidc": {"submission_via_relay": True}}},
)
variables = {
"MAIL_PROVIDER": "web-app-mailprov",
"group_names": ["web-app-mailprov", "web-app-keycloak"],
"groups": {"web-app-mailprov": ["host1"], "web-app-keycloak": ["host1"]},
"TLS_ENABLED": True,
"inventory_hostname": "host1",
}
result = self.lookup.run(
[], variables=variables, roles_dir=str(self._tmp / "roles")
)[0]
self.assertEqual(result["port"], 25)
self.assertFalse(result["auth"])
self.assertTrue(result["start_tls"])
def test_sso_relay_inactive_without_keycloak(self) -> None:
# Same provider, but Keycloak is not deployed: keep authenticated 465.
_write_role_config(
self._tmp,
"web-app-mailprov",
{"sso": {"oidc": {"submission_via_relay": True}}},
)
variables = {
"MAIL_PROVIDER": "web-app-mailprov",
"group_names": ["web-app-mailprov"],
"groups": {"web-app-mailprov": ["host1"]},
"TLS_ENABLED": True,
"inventory_hostname": "host1",
}
result = self.lookup.run(
[], variables=variables, roles_dir=str(self._tmp / "roles")
)[0]
self.assertEqual(result["port"], 465)
self.assertTrue(result["auth"])
self.assertFalse(result["start_tls"])
def test_sso_relay_inactive_when_sso_enabled_pinned_false(self) -> None:
# A variant pins the provider's sso.enabled to a literal false while
# submission_via_relay stays true (role default) and Keycloak is still
# deployed: the provider keeps password auth, so no relay mode.
_write_role_config(
self._tmp,
"web-app-mailprov",
{"sso": {"enabled": False, "oidc": {"submission_via_relay": True}}},
)
variables = {
"MAIL_PROVIDER": "web-app-mailprov",
"group_names": ["web-app-mailprov", "web-app-keycloak"],
"groups": {"web-app-mailprov": ["host1"], "web-app-keycloak": ["host1"]},
"TLS_ENABLED": True,
"inventory_hostname": "host1",
}
result = self.lookup.run(
[], variables=variables, roles_dir=str(self._tmp / "roles")
)[0]
self.assertEqual(result["port"], 465)
self.assertTrue(result["auth"])
self.assertFalse(result["start_tls"])
def test_a_provider_without_plaintext_submission_falls_back_to_its_mx(
self,
) -> None:
"""Stalwart's default config binds the implicit-TLS client ports and the
MX, never 587, so the onion path (which drops TLS) has no submission
listener to reach. The port follows what the provider declares, and AUTH
drops with it: an MX offers none, and announcing it aborts msmtp with
EX_UNAVAILABLE before a message is ever sent."""
_write_role_config(
self._tmp,
"web-app-mailprov",
{"mailprov": {"ports": {"public": {"smtp": 25, "smtps": 465}}}},
)
variables = {
"MAIL_PROVIDER": "web-app-mailprov",
"group_names": ["web-app-mailprov"],
"groups": {"web-app-mailprov": ["host1"]},
"SYSTEM_EMAIL_HOST": "mail.x.onion",
"TLS_ENABLED": True,
"DOMAIN_PRIMARY": "x.onion",
"inventory_hostname": "host1",
}
result = self.lookup.run(
[], variables=variables, roles_dir=str(self._tmp / "roles")
)[0]
self.assertFalse(result["tls"])
self.assertEqual(result["port"], 25)
self.assertFalse(result["auth"])
self.assertEqual(result["auth_mechanism"], "off")
def test_a_provider_declaring_submission_keeps_the_authenticated_path(
self,
) -> None:
"""Mailu binds 587, so the same onion target stays on plaintext
submission with AUTH PLAIN. The MX fallback above must not reach a
provider that declares a submission listener."""
_write_role_config(
self._tmp,
"web-app-mailprov",
{
"mailprov": {
"ports": {"public": {"smtp": 25, "smtps": 465, "submission": 587}}
}
},
)
variables = {
"MAIL_PROVIDER": "web-app-mailprov",
"group_names": ["web-app-mailprov"],
"groups": {"web-app-mailprov": ["host1"]},
"SYSTEM_EMAIL_HOST": "mail.x.onion",
"TLS_ENABLED": True,
"DOMAIN_PRIMARY": "x.onion",
"inventory_hostname": "host1",
}
result = self.lookup.run(
[], variables=variables, roles_dir=str(self._tmp / "roles")
)[0]
self.assertFalse(result["tls"])
self.assertEqual(result["port"], 587)
self.assertTrue(result["auth"])
self.assertEqual(result["auth_mechanism"], "plain")
def test_sso_relay_active_with_untemplated_enabled_gate(self) -> None:
# The role default gates sso.enabled on group_names (raw Jinja here);
# the guard must not treat the untemplated string as false.
_write_role_config(
self._tmp,
"web-app-mailprov",
{
"sso": {
"enabled": "{{ 'web-app-keycloak' in group_names }}",
"oidc": {"submission_via_relay": True},
}
},
)
variables = {
"MAIL_PROVIDER": "web-app-mailprov",
"group_names": ["web-app-mailprov", "web-app-keycloak"],
"groups": {"web-app-mailprov": ["host1"], "web-app-keycloak": ["host1"]},
"TLS_ENABLED": True,
"inventory_hostname": "host1",
}
result = self.lookup.run(
[], variables=variables, roles_dir=str(self._tmp / "roles")
)[0]
self.assertEqual(result["port"], 25)
self.assertFalse(result["auth"])
self.assertTrue(result["start_tls"])
def test_computed_defaults_are_templated(self) -> None:
self.lookup._templar = _DummyTemplar(
{"DOMAIN_PRIMARY_RESOLVED": "mail.example.org"}
)
variables = {
"DOMAIN_PRIMARY": "{{ DOMAIN_PRIMARY_RESOLVED }}",
"inventory_hostname": "host1",
}
result = self.lookup.run([], variables=variables)[0]
self.assertEqual(result["domain"], "mail.example.org")
class TestProviderOnionIsClusterWide(unittest.TestCase):
"""In swarm only the manager is in svc-net-tor, so a worker renders the
provider's clearnet name while the provider (onion-primary on the
manager) never deployed a certificate for it. TLS must follow the
provider host's view, not the worker's."""
ONION = "a" * 56 + ".onion"
TOR_ON = "{{ 'svc-net-tor' in group_names }}"
STALWART_PORTS: ClassVar[dict[str, int]] = {"smtp": 25, "smtps": 465}
MAILU_PORTS: ClassVar[dict[str, int]] = {
"smtp": 25,
"smtps": 465,
"submission": 587,
}
@classmethod
def setUpClass(cls) -> None:
_reset_cache_for_tests()
cls.addClassCleanup(_reset_cache_for_tests)
def setUp(self) -> None:
from ansible.parsing.dataloader import DataLoader
from ansible.template import Templar
cache_base._reset()
_reset_cache_for_tests()
self.lookup = LookupModule()
self.lookup._templar = Templar(loader=DataLoader(), variables={})
self._cwd = str(Path.cwd())
self._tmpdir = tempfile.TemporaryDirectory()
self._tmp = Path(self._tmpdir.name)
(self._tmp / "roles").mkdir(parents=True, exist_ok=True)
os.chdir(self._tmp)
self._tokens_store_patcher = patch.object(
cache_users, "_load_store_users", return_value={}
)
self._tokens_store_patcher.start()
def tearDown(self) -> None:
self._tokens_store_patcher.stop()
os.chdir(self._cwd)
self._tmpdir.cleanup()
def _seed(self, *, ports, node=None, provider_tor=None) -> None:
"""Write the tor provider and the mail provider roles.
Args:
ports: the mail provider's published SMTP ports.
node: svc-net-tor's node onion; defaults to ``ONION``.
provider_tor: the mail provider's ``tor`` block; defaults to
enabled on svc-net-tor hosts, inheriting exclusive/primary.
"""
_write_role_config(
self._tmp,
"svc-net-tor",
{
"tor": {
"enabled": True,
"shared": True,
"exclusive": True,
"primary": True,
"node": self.ONION if node is None else node,
}
},
)
_write_role_config(
self._tmp,
"web-app-mailprov",
{
"mailprov": {"ports": {"public": ports}},
"tor": provider_tor or {"enabled": self.TOR_ON},
},
)
domains = self._tmp / "roles" / "web-app-mailprov" / ROLE_FILE_META_DOMAINS
domains.write_text(
dump_yaml_str({"canonical": {"mail": "mail.x.test"}, "aliases": []}),
encoding="utf-8",
)
def _worker(self, *, tor_hosts=("mgr",)) -> dict:
return {
"MAIL_PROVIDER": "web-app-mailprov",
"group_names": ["svc-swarm-node"],
"groups": {
"web-app-mailprov": ["mgr"],
"svc-net-tor": list(tor_hosts),
},
"hostvars": {
"mgr": {"group_names": ["web-app-mailprov", "svc-net-tor"]},
"other": {"group_names": ["svc-net-tor"]},
},
"SYSTEM_EMAIL_HOST": "mail.x.test",
"TLS_ENABLED": True,
"DOMAIN_PRIMARY": "x.test",
"inventory_hostname": "wrk",
}
def _run(self, variables: dict) -> dict:
return self.lookup.run(
[], variables=variables, roles_dir=str(self._tmp / "roles")
)[0]
def test_worker_of_an_onion_stalwart_drops_tls_and_auth(self) -> None:
self._seed(ports=self.STALWART_PORTS)
result = self._run(self._worker())
self.assertFalse(result["tls"])
self.assertFalse(result["start_tls"])
self.assertFalse(result["auth"])
self.assertEqual(result["port"], 25)
def test_worker_of_an_onion_mailu_keeps_plain_submission(self) -> None:
self._seed(ports=self.MAILU_PORTS)
result = self._run(self._worker())
self.assertFalse(result["tls"])
self.assertEqual(result["port"], 587)
self.assertTrue(result["auth"])
self.assertEqual(result["auth_mechanism"], "plain")
def test_tor_on_another_host_keeps_tls(self) -> None:
self._seed(ports=self.STALWART_PORTS)
result = self._run(self._worker(tor_hosts=("other",)))
self.assertTrue(result["tls"])
self.assertEqual(result["port"], 465)
def test_dual_stack_provider_keeps_tls(self) -> None:
self._seed(
ports=self.STALWART_PORTS,
provider_tor={"enabled": self.TOR_ON, "exclusive": False, "primary": False},
)
result = self._run(self._worker())
self.assertTrue(result["tls"])
def test_empty_tor_node_keeps_tls(self) -> None:
self._seed(ports=self.STALWART_PORTS, node="")
result = self._run(self._worker())
self.assertTrue(result["tls"])
class _DummyTemplar:
def __init__(self, available_variables: dict[str, str]) -> None:
self.available_variables = available_variables
def template(self, value, fail_on_undefined=False):
if isinstance(value, str):
rendered = value
for key, replacement in self.available_variables.items():
rendered = rendered.replace(f"{{{{ {key} }}}}", str(replacement))
return rendered
return value
if __name__ == "__main__":
unittest.main()
Reference in new issue View Git Blame Copy Permalink