454 lines
19 KiB
Python
454 lines
19 KiB
Python
# 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()
|