# SPDX-FileCopyrightText: 2026 Eric Ireland # SPDX-License-Identifier: GPL-3.0-or-later import importlib.util from importlib.machinery import SourceFileLoader from contextlib import contextmanager import os from pathlib import Path import stat import sys import tempfile import textwrap import unittest from unittest import mock PROJECT_ROOT = Path(__file__).resolve().parents[1] SOURCE = PROJECT_ROOT / "vmailctl" SPEC = importlib.util.spec_from_loader( "vmailctl", SourceFileLoader("vmailctl", os.fspath(SOURCE)), ) vmailctl = importlib.util.module_from_spec(SPEC) assert SPEC.loader is not None sys.modules[SPEC.name] = vmailctl SPEC.loader.exec_module(vmailctl) @contextmanager def relaxed_transaction_paths(): @contextmanager def unlocked(_path): yield with ( mock.patch.object(vmailctl, "exclusive_lock", unlocked), mock.patch.object(vmailctl, "ensure_secure_backup_directory"), ): yield class ParsingTests(unittest.TestCase): def test_normalize_localpart(self): self.assertEqual( vmailctl.normalize_localpart("Alice.Smith", "example.com"), ("alice.smith", "alice.smith@example.com"), ) self.assertEqual( vmailctl.normalize_localpart("alice@example.com", "example.com"), ("alice", "alice@example.com"), ) def test_rejects_unsafe_localparts(self): for value in ("alice+tag", ".alice", "alice.", "alice..smith", "alice@example.net", "a/b"): with self.subTest(value=value), self.assertRaises(vmailctl.VmailError): vmailctl.normalize_localpart(value, "example.com") def test_parse_passwd(self): text = ( "# comment\n" "alice@example.com:{ARGON2ID}hash:5000:5000::" "/var/mail/vhosts/example.com/alice::userdb_mail_path=~/Maildir\n" ) parsed = vmailctl.parse_passwd(text, "users") self.assertEqual(list(parsed), ["alice@example.com"]) self.assertEqual(parsed["alice@example.com"].fields[2:4], ("5000", "5000")) def test_parse_passwd_rejects_duplicate(self): line = "alice@example.com:hash:5000:5000::/mail/alice::userdb_mail_path=~/Maildir\n" with self.assertRaises(vmailctl.VmailError): vmailctl.parse_passwd(line + line, "users") def test_parse_map(self): parsed = vmailctl.parse_map( "# comment\nalice@example.com example.com/alice/Maildir/\n", "vmailbox", ) self.assertEqual(parsed["alice@example.com"][1], "example.com/alice/Maildir/") def test_append_line_repairs_missing_newline(self): self.assertEqual(vmailctl.append_line("one", "two"), "one\ntwo\n") def test_rejects_invalid_domains(self): for value in ( ".example.com", "example..com", "-example.com", "example-.com", "localhost", "example_com", ): with self.subTest(value=value), self.assertRaises(vmailctl.VmailError): vmailctl.validate_domain(value, "test") def test_rejects_unsafe_and_duplicate_folder_names(self): for folders in ( ("-A",), ("../Archive",), ("Junk\nInjected",), ("Archive", "Archive"), ): with self.subTest(folders=folders), self.assertRaises(vmailctl.VmailError): vmailctl.validate_folders(folders, "test") class FileSafetyTests(unittest.TestCase): @staticmethod def metadata(mode, uid=0, gid=0): return os.stat_result((mode, 0, 0, 1, uid, gid, 0, 0, 0, 0)) def test_trusted_file_metadata_rules(self): safe = self.metadata(stat.S_IFREG | 0o644) vmailctl.validate_regular_metadata(safe, "safe", private=False) for metadata in ( self.metadata(stat.S_IFREG | 0o666), self.metadata(stat.S_IFREG | 0o644, uid=1000, gid=1000), self.metadata(stat.S_IFLNK | 0o777), ): with self.subTest(mode=metadata.st_mode), self.assertRaises(vmailctl.VmailError): vmailctl.validate_regular_metadata(metadata, "unsafe", private=False) def test_private_and_executable_metadata_rules(self): with self.assertRaises(vmailctl.VmailError): vmailctl.validate_regular_metadata( self.metadata(stat.S_IFREG | 0o640), "private", private=True, ) with self.assertRaises(vmailctl.VmailError): vmailctl.validate_regular_metadata( self.metadata(stat.S_IFREG | 0o644), "command", private=False, executable=True, ) def test_replace_text_preserves_metadata(self): with tempfile.TemporaryDirectory() as temporary: path = Path(temporary) / "users" path.write_text("old\n", encoding="utf-8") path.chmod(0o640) vmailctl.replace_text(path, "new\n") self.assertEqual(path.read_text(encoding="utf-8"), "new\n") self.assertEqual(stat.S_IMODE(path.stat().st_mode), 0o640) def test_backup_restore_text_and_database(self): with tempfile.TemporaryDirectory() as temporary: root = Path(temporary) source = root / "virtual" database = root / "virtual.db" source.write_text("old source\n", encoding="utf-8") database.write_bytes(b"old database") settings = mock.Mock() settings.backup_dir = root / "backups" with mock.patch.object(vmailctl, "ensure_secure_backup_directory"): settings.backup_dir.mkdir() backup = vmailctl.BackupSet(settings, "test", (source, database)) source.write_text("new source\n", encoding="utf-8") database.write_bytes(b"new database") backup.restore() self.assertEqual(source.read_text(encoding="utf-8"), "old source\n") self.assertEqual(database.read_bytes(), b"old database") def test_malformed_config_is_a_controlled_error(self): with tempfile.TemporaryDirectory() as temporary: path = Path(temporary) / "vmailctl.conf" path.write_text("[broken\n", encoding="utf-8") with ( mock.patch.object(vmailctl, "validate_regular_metadata"), self.assertRaises(vmailctl.VmailError), ): vmailctl.Settings.load(path) def test_lock_file_is_private_and_uses_strict_metadata_call(self): with tempfile.TemporaryDirectory() as temporary: path = Path(temporary) / "vmailctl.lock" checked = [] def strict_check(metadata, label, *, private, executable=False): checked.append((metadata, label, private, executable)) with ( mock.patch.object(vmailctl, "validate_secure_directory"), mock.patch.object(vmailctl, "validate_regular_metadata", strict_check), vmailctl.exclusive_lock(path), ): self.assertTrue(path.exists()) self.assertEqual(len(checked), 1) self.assertTrue(checked[0][2]) self.assertEqual(stat.S_IMODE(path.stat().st_mode), 0o600) class PasswordTests(unittest.TestCase): def test_generated_password_length_and_alphabet(self): password = vmailctl.generate_password() self.assertEqual(len(password), 24) self.assertTrue(set(password) <= set(vmailctl.GENERATED_PASSWORD_ALPHABET)) @mock.patch.object(vmailctl, "run") def test_hash_password_uses_stdin_not_argv(self, mocked_run): mocked_run.return_value = mock.Mock(stdout="{ARGON2ID}result\n") settings = mock.Mock() settings.doveadm = Path("/usr/bin/doveadm") settings.hash_scheme = "ARGON2ID" result = vmailctl.hash_password(settings, "secret-value") self.assertEqual(result, "{ARGON2ID}result") argv = mocked_run.call_args.args[0] self.assertNotIn("secret-value", [os.fspath(item) for item in argv]) self.assertEqual(mocked_run.call_args.kwargs["input_text"], "secret-value\nsecret-value\n") class ReleaseMetadataTests(unittest.TestCase): def test_version_is_consistent_across_public_documentation(self): version = vmailctl.VERSION readme = (PROJECT_ROOT / "README.md").read_text(encoding="utf-8") changelog = (PROJECT_ROOT / "CHANGELOG.md").read_text(encoding="utf-8") manual = (PROJECT_ROOT / "man" / "vmailctl.8").read_text(encoding="utf-8") self.assertIn(f"`{version}` is a public beta", readme) self.assertIn(f"## [{version}]", changelog) self.assertIn(f'"vmailctl {version}"', manual) class TransactionLogicTests(unittest.TestCase): def fixture_settings(self, root: Path): users = root / "virtual-users" mailboxes = root / "vmailbox" aliases = root / "virtual" users.write_text( f"alice@example.com:{{ARGON2ID}}hash:{os.getuid()}:{os.getgid()}::" f"{root}/mail/example.com/alice::userdb_mail_path=~/Maildir\n", encoding="utf-8", ) mailboxes.write_text( "alice@example.com example.com/alice/Maildir/\n", encoding="utf-8", ) aliases.write_text("shopping@example.com alice@example.com\n", encoding="utf-8") for path in (users, mailboxes, aliases): path.chmod(0o640) (root / "mail" / "example.com" / "alice" / "Maildir").mkdir(parents=True) mailboxes.with_name(mailboxes.name + ".db").write_bytes(b"old mailbox db") aliases.with_name(aliases.name + ".db").write_bytes(b"old alias db") fake_postmap = root / "postmap" fake_postmap.write_text( textwrap.dedent( """\ #!/usr/bin/env python3 from pathlib import Path import sys if sys.argv[1] == "-q": key = sys.argv[2] source = Path(sys.argv[3].removeprefix("hash:")) for raw in source.read_text().splitlines(): fields = raw.split() if len(fields) == 2 and fields[0].lower() == key.lower(): print(fields[1]) break else: Path(sys.argv[1] + ".db").write_bytes(b"compiled map") """ ), encoding="utf-8", ) fake_postmap.chmod(0o755) fake_doveadm = root / "doveadm" fake_doveadm.write_text( "#!/bin/sh\n" "if [ \"$1\" = pw ]; then printf '%s\\n' '{ARGON2ID}testhash'; fi\n" "exit 0\n", encoding="utf-8", ) fake_doveadm.chmod(0o755) (root / "backups").mkdir() return vmailctl.Settings( domain="example.com", dovecot_users=users, postfix_mailboxes=mailboxes, postfix_aliases=aliases, mail_root=root / "mail", uid=os.getuid(), gid=os.getgid(), hash_scheme="ARGON2ID", folders=("Archive", "Sent"), backup_dir=root / "backups", lock_file=root / "vmailctl.lock", postmap=fake_postmap, postfix=Path("/bin/true"), doveadm=fake_doveadm, doveconf=Path("/bin/true"), directory_mode=0o700, ) def test_dry_run_add_does_not_prompt_or_write(self): with tempfile.TemporaryDirectory() as temporary: settings = self.fixture_settings(Path(temporary)) before = settings.dovecot_users.read_bytes() with mock.patch.object(vmailctl, "prompt_password") as prompt: vmailctl.mailbox_add(settings, "bob", True, False) prompt.assert_not_called() self.assertEqual(settings.dovecot_users.read_bytes(), before) self.assertFalse(settings.mailbox_home("bob").exists()) def test_alias_requires_real_mailbox_target(self): with tempfile.TemporaryDirectory() as temporary: settings = self.fixture_settings(Path(temporary)) with self.assertRaises(vmailctl.VmailError): vmailctl.alias_add(settings, "newalias", "missing", True) def test_new_passwd_line_matches_live_schema(self): with tempfile.TemporaryDirectory() as temporary: settings = self.fixture_settings(Path(temporary)) line = vmailctl.new_passwd_line( settings, "bob", "bob@example.com", "{ARGON2ID}hash", ) fields = line.split(":") self.assertEqual(len(fields), 8) self.assertEqual(fields[2:4], [str(os.getuid()), str(os.getgid())]) self.assertEqual(fields[7], "userdb_mail_path=~/Maildir") def test_full_mailbox_add_transaction(self): with tempfile.TemporaryDirectory() as temporary: settings = self.fixture_settings(Path(temporary)) with ( relaxed_transaction_paths(), mock.patch.object(vmailctl, "generate_password", return_value="StrongGeneratedPassword1"), ): vmailctl.mailbox_add(settings, "bob", False, True) passwd = vmailctl.parse_passwd( settings.dovecot_users.read_text(encoding="utf-8"), "users", ) mailboxes = vmailctl.parse_map( settings.postfix_mailboxes.read_text(encoding="utf-8"), "vmailbox", ) self.assertIn("bob@example.com", passwd) self.assertEqual( mailboxes["bob@example.com"][1], "example.com/bob/Maildir/", ) self.assertTrue((settings.mailbox_home("bob") / "Maildir" / "cur").is_dir()) self.assertNotIn( "StrongGeneratedPassword1", settings.dovecot_users.read_text(encoding="utf-8"), ) def test_full_alias_add_transaction(self): with tempfile.TemporaryDirectory() as temporary: settings = self.fixture_settings(Path(temporary)) with relaxed_transaction_paths(): vmailctl.alias_add(settings, "orders", "alice", False) aliases = vmailctl.parse_map( settings.postfix_aliases.read_text(encoding="utf-8"), "virtual", ) self.assertEqual(aliases["orders@example.com"][1], "alice@example.com") def test_mailbox_add_rolls_back_and_quarantines_home(self): with tempfile.TemporaryDirectory() as temporary: settings = self.fixture_settings(Path(temporary)) old_users = settings.dovecot_users.read_bytes() old_mailboxes = settings.postfix_mailboxes.read_bytes() with ( relaxed_transaction_paths(), mock.patch.object(vmailctl, "generate_password", return_value="StrongGeneratedPassword1"), mock.patch.object( vmailctl, "verify_postfix", side_effect=vmailctl.VmailError("injected validation failure"), ), self.assertRaises(vmailctl.VmailError), ): vmailctl.mailbox_add(settings, "bob", False, True) self.assertEqual(settings.dovecot_users.read_bytes(), old_users) self.assertEqual(settings.postfix_mailboxes.read_bytes(), old_mailboxes) self.assertFalse(settings.mailbox_home("bob").exists()) quarantined = list(settings.backup_dir.glob("*/rolled-back-mailhome")) self.assertEqual(len(quarantined), 1) def test_mailbox_add_interrupt_rolls_back_and_quarantines_home(self): with tempfile.TemporaryDirectory() as temporary: settings = self.fixture_settings(Path(temporary)) old_users = settings.dovecot_users.read_bytes() old_mailboxes = settings.postfix_mailboxes.read_bytes() with ( relaxed_transaction_paths(), mock.patch.object(vmailctl, "generate_password", return_value="StrongGeneratedPassword1"), mock.patch.object(vmailctl, "verify_dovecot", side_effect=KeyboardInterrupt()), self.assertRaises(KeyboardInterrupt), ): vmailctl.mailbox_add(settings, "bob", False, True) self.assertEqual(settings.dovecot_users.read_bytes(), old_users) self.assertEqual(settings.postfix_mailboxes.read_bytes(), old_mailboxes) self.assertFalse(settings.mailbox_home("bob").exists()) quarantined = list(settings.backup_dir.glob("*/rolled-back-mailhome")) self.assertEqual(len(quarantined), 1) def test_password_interrupt_rolls_back(self): with tempfile.TemporaryDirectory() as temporary: settings = self.fixture_settings(Path(temporary)) old_users = settings.dovecot_users.read_bytes() with ( relaxed_transaction_paths(), mock.patch.object(vmailctl, "generate_password", return_value="StrongGeneratedPassword1"), mock.patch.object(vmailctl, "verify_dovecot", side_effect=KeyboardInterrupt()), self.assertRaises(KeyboardInterrupt), ): vmailctl.mailbox_password(settings, "alice", False, True) self.assertEqual(settings.dovecot_users.read_bytes(), old_users) def test_alias_interrupt_rolls_back(self): with tempfile.TemporaryDirectory() as temporary: settings = self.fixture_settings(Path(temporary)) old_aliases = settings.postfix_aliases.read_bytes() old_database = Path(str(settings.postfix_aliases) + ".db").read_bytes() with ( relaxed_transaction_paths(), mock.patch.object(vmailctl, "verify_postfix", side_effect=KeyboardInterrupt()), self.assertRaises(KeyboardInterrupt), ): vmailctl.alias_add(settings, "orders", "alice", False) self.assertEqual(settings.postfix_aliases.read_bytes(), old_aliases) self.assertEqual( Path(str(settings.postfix_aliases) + ".db").read_bytes(), old_database, ) def test_consistent_legacy_map_overlap_is_not_an_audit_error(self): with tempfile.TemporaryDirectory() as temporary: settings = self.fixture_settings(Path(temporary)) with settings.postfix_mailboxes.open("a", encoding="utf-8") as handle: handle.write("postmaster@example.com example.com/alice/Maildir/\n") with settings.postfix_aliases.open("a", encoding="utf-8") as handle: handle.write("postmaster@example.com alice@example.com\n") os.chown(settings.mailbox_home("alice") / "Maildir", settings.uid, settings.gid) with ( mock.patch.object(vmailctl, "query_map") as query, mock.patch.object(vmailctl, "run", return_value=mock.Mock(returncode=0)), ): def lookup(_settings, path, key): parsed = vmailctl.parse_map(path.read_text(encoding="utf-8"), str(path)) return parsed[key][1] query.side_effect = lookup self.assertEqual(vmailctl.audit(settings), 0) if __name__ == "__main__": unittest.main()