Mario Fetka
466 lines
20 KiB
Python
466 lines
20 KiB
Python
# /****************************************************************************
|
|
# * <Novell-copyright>
|
|
# * Copyright (c) 2001 Novell, Inc. All Rights Reserved.
|
|
# *
|
|
# * This program is free software; you can redistribute it and/or
|
|
# * modify it under the terms of version 2 of the GNU General Public License
|
|
# * as published by the Free Software Foundation.
|
|
# *
|
|
# * This program is distributed in the hope that it will be useful,
|
|
# * but WITHOUT ANY WARRANTY; without even the implied warranty of
|
|
# * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
|
# * GNU General Public License for more details.
|
|
# *
|
|
# * You should have received a copy of the GNU General Public License
|
|
# * along with this program; if not, contact Novell, Inc.
|
|
# *
|
|
# * To contact Novell about this file by physical or electronic mail, you
|
|
# * may find current contact information at www.novell.com.
|
|
# * </Novell-copyright>
|
|
# ****************************************************************************/
|
|
|
|
from __future__ import annotations
|
|
|
|
import copy
|
|
import json
|
|
import os
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
from unittest import mock
|
|
|
|
SOURCE = Path(__file__).resolve().parents[4]
|
|
sys.path.insert(0, str(SOURCE / "src" / "libs" / "python"))
|
|
|
|
from bongo.configuration.model import ( # noqa: E402
|
|
Scenario, STORE_DOCUMENTS, apply_scenario, json_bytes, merge_defaults,
|
|
)
|
|
from bongo.configuration.probes import probe_clamav, probe_spamd # noqa: E402
|
|
from bongo.configuration.storage import ( # noqa: E402
|
|
ConfigurationError, ConfigurationPaths, ConfigurationStore,
|
|
)
|
|
|
|
|
|
class FakeStore(ConfigurationStore):
|
|
def __init__(self, paths):
|
|
super().__init__(paths)
|
|
scenario = Scenario(
|
|
hostname="mail.example.test", bind_address="192.0.2.10",
|
|
domains=["example.test"], admin_password="0123456789ab",
|
|
web_mode="direct",
|
|
web_public_url="http://mail.example.test:8080",
|
|
)
|
|
self.live = apply_scenario(self.templates, scenario)
|
|
self.aliases = {
|
|
"default_config": copy.deepcopy(self.alias_template),
|
|
"example.test": copy.deepcopy(self.alias_template),
|
|
}
|
|
self.domain_aliases = {}
|
|
self.username_mappings = {}
|
|
self.fail_name = None
|
|
|
|
def read_store_document(self, name):
|
|
value = copy.deepcopy(self.live[name])
|
|
return value, json_bytes(value)
|
|
|
|
def write_store_document(self, name, content):
|
|
if name == self.fail_name:
|
|
self.fail_name = None
|
|
raise ConfigurationError("injected Store failure")
|
|
self.live[name] = json.loads(content)
|
|
|
|
def read_store_path(self, name):
|
|
if name.startswith("aliases/"):
|
|
value = {
|
|
"domainalias": self.domain_aliases.get(
|
|
name.split("/", 1)[1], ""),
|
|
"aliases": copy.deepcopy(
|
|
self.aliases[name.split("/", 1)[1]]["aliases"]),
|
|
"username-mapping": self.username_mappings.get(
|
|
name.split("/", 1)[1], 0),
|
|
}
|
|
return value, json_bytes(value)
|
|
return self.read_store_document(name)
|
|
|
|
|
|
class FakeSocket:
|
|
def __init__(self, response):
|
|
self.response = response
|
|
self.sent = b""
|
|
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, *_args):
|
|
return False
|
|
|
|
def settimeout(self, _timeout):
|
|
pass
|
|
|
|
def sendall(self, content):
|
|
self.sent += content
|
|
|
|
def recv(self, _maximum):
|
|
response, self.response = self.response, b""
|
|
return response
|
|
|
|
|
|
class ConfigurationStorageTest(unittest.TestCase):
|
|
def setUp(self):
|
|
self.temporary = tempfile.TemporaryDirectory()
|
|
root = Path(self.temporary.name)
|
|
self.paths = ConfigurationPaths(
|
|
templates=SOURCE / "src" / "apps" / "config" / "config",
|
|
config_dir=root / "etc" / "config.d",
|
|
aliases_dir=root / "etc" / "aliases.d",
|
|
ssl_dir=root / "etc" / "ssl.d",
|
|
dkim_dir=root / "etc" / "dkim.d",
|
|
dmarc_dir=root / "etc" / "dmarc.d",
|
|
acme_dir=root / "etc" / "acme.d",
|
|
acme_provider_config=root / "etc" / "acme.d" / "providers",
|
|
legacy_certificate=root / "var" / "lib" / "bongo" / "dbf" /
|
|
"osslcert.pem",
|
|
legacy_key=root / "var" / "lib" / "bongo" / "dbf" /
|
|
"osslpriv.pem",
|
|
legacy_web=root / "etc" / "web",
|
|
legacy_web_json=root / "etc" / "bongo-web.json",
|
|
legacy_authentication=root / "etc" / "bongo.conf",
|
|
migration_backup=root / "etc" / "migration-backup" / "pre-0.7",
|
|
admin=root / "sbin" / "bongo-admin",
|
|
config=root / "sbin" / "bongo-config",
|
|
lock=root / "run" / "bongo-config.lock",
|
|
)
|
|
self.store = FakeStore(self.paths)
|
|
|
|
def tearDown(self):
|
|
self.temporary.cleanup()
|
|
|
|
def test_save_writes_local_documents(self):
|
|
configurations = copy.deepcopy(self.store.live)
|
|
configurations["web"]["port"] = 8088
|
|
configurations["authentication"]["AuthLdapOfflineCacheHours"] = 72
|
|
self.store.save(configurations)
|
|
self.assertEqual(
|
|
json.loads(self.paths.document("web").read_text())["port"], 8088)
|
|
self.assertEqual(json.loads(
|
|
self.paths.document("authentication").read_text()
|
|
)["AuthLdapOfflineCacheHours"], 72)
|
|
|
|
def test_save_canonicalizes_internationalized_configuration_domains(self):
|
|
configurations = copy.deepcopy(self.store.live)
|
|
configurations["global"]["hostname"] = "mail.bücher.example"
|
|
configurations["global"]["tls_hostnames"] = [
|
|
"mail.bücher.example"]
|
|
configurations["queue"]["domains"] = ["bücher.example"]
|
|
configurations["queue"]["hosteddomains"] = ["bücher.example"]
|
|
self.store.save(configurations)
|
|
global_config = json.loads(
|
|
self.paths.document("global").read_text())
|
|
queue = json.loads(self.paths.document("queue").read_text())
|
|
self.assertEqual(global_config["hostname"],
|
|
"mail.xn--bcher-kva.example")
|
|
self.assertEqual(queue["domains"], ["xn--bcher-kva.example"])
|
|
|
|
def test_install_is_repeatable_and_preserves_existing_aliases(self):
|
|
configurations = copy.deepcopy(self.store.live)
|
|
invocations = []
|
|
|
|
def completed(command, *, pass_fds, check):
|
|
self.assertFalse(check)
|
|
password_fd, bundle_fd = pass_fds
|
|
os.lseek(password_fd, 0, os.SEEK_SET)
|
|
os.lseek(bundle_fd, 0, os.SEEK_SET)
|
|
password = os.read(password_fd, 4096)
|
|
bundle = json.loads(os.read(bundle_fd, 1024 * 1024))
|
|
invocations.append((list(command), password, bundle))
|
|
return subprocess.CompletedProcess(command, 0)
|
|
|
|
with mock.patch("subprocess.run", side_effect=completed):
|
|
self.store.install(
|
|
configurations, "mail.example.test", "192.0.2.10",
|
|
["example.test"], "0123456789ab", ["mail.example.test"])
|
|
first_documents = {
|
|
name: self.paths.document(name).read_bytes()
|
|
for name in STORE_DOCUMENTS
|
|
}
|
|
aliases = json.loads(
|
|
(self.paths.aliases_dir / "default_config").read_text())
|
|
aliases["aliases"]["abuse"] = "security"
|
|
(self.paths.aliases_dir / "default_config").write_bytes(
|
|
json_bytes(aliases))
|
|
|
|
self.store.install(
|
|
configurations, "mail.example.test", "192.0.2.10",
|
|
["example.test"], "0123456789ab", ["mail.example.test"])
|
|
|
|
self.assertEqual(len(invocations), 2)
|
|
self.assertEqual(invocations[0], invocations[1])
|
|
self.assertEqual(invocations[0][1], b"0123456789ab")
|
|
self.assertEqual(
|
|
set(invocations[0][2]), STORE_DOCUMENTS)
|
|
self.assertEqual(first_documents, {
|
|
name: self.paths.document(name).read_bytes()
|
|
for name in STORE_DOCUMENTS
|
|
})
|
|
aliases = json.loads(
|
|
(self.paths.aliases_dir / "default_config").read_text())
|
|
self.assertEqual(aliases["aliases"]["abuse"], "security")
|
|
self.assertEqual(aliases["aliases"]["dmarc-reports"], "admin")
|
|
|
|
def test_worker_defaults_append_new_registered_jobs(self):
|
|
configurations = copy.deepcopy(self.store.live)
|
|
configurations["worker"]["jobs"] = [
|
|
job for job in configurations["worker"]["jobs"]
|
|
if job["name"] != "scanner-health"
|
|
]
|
|
result = merge_defaults(configurations, self.store.templates)
|
|
self.assertEqual(
|
|
[job["name"] for job in result["worker"]["jobs"]],
|
|
["acme-renew", "tlsrpt-delivery", "scanner-health"],
|
|
)
|
|
|
|
def test_migration_moves_historical_files_and_writes_all_documents(self):
|
|
self.paths.legacy_web_json.parent.mkdir(parents=True)
|
|
self.paths.legacy_web_json.write_bytes(json_bytes(self.store.live["web"]))
|
|
self.paths.legacy_authentication.write_text(
|
|
"AuthLdapEnabled = false\nAuthLdapOfflineCacheHours = 72\n")
|
|
moved = self.store.migrate_legacy()
|
|
self.assertEqual(len(moved), 2)
|
|
self.assertFalse(self.paths.legacy_web_json.exists())
|
|
self.assertFalse(self.paths.legacy_authentication.exists())
|
|
self.assertTrue((self.paths.migration_backup / "bongo-web.json").exists())
|
|
self.assertEqual(
|
|
set(path.name for path in self.paths.config_dir.iterdir()),
|
|
set(STORE_DOCUMENTS))
|
|
self.assertEqual(
|
|
set(path.name for path in self.paths.aliases_dir.iterdir()),
|
|
{"default_config"})
|
|
self.assertEqual(json.loads(
|
|
(self.paths.aliases_dir / "default_config").read_text()
|
|
)["aliases"]["dmarc-reports"], "admin")
|
|
self.assertEqual(json.loads(
|
|
self.paths.document("authentication").read_text()
|
|
)["AuthLdapOfflineCacheHours"], 72)
|
|
|
|
def test_migration_refuses_ambiguous_historical_web_files(self):
|
|
self.paths.legacy_web.parent.mkdir(parents=True)
|
|
content = json_bytes(self.store.live["web"])
|
|
self.paths.legacy_web.write_bytes(content)
|
|
self.paths.legacy_web_json.write_bytes(content)
|
|
with self.assertRaisesRegex(ConfigurationError, "multiple historical"):
|
|
self.store.migrate_legacy()
|
|
|
|
def test_migration_moves_complete_legacy_tls_pair(self):
|
|
self.paths.legacy_certificate.parent.mkdir(parents=True)
|
|
self.paths.legacy_certificate.write_text("certificate")
|
|
self.paths.legacy_key.write_text("private key")
|
|
moved = self.store.migrate_legacy()
|
|
self.assertIn((self.paths.legacy_certificate,
|
|
self.paths.ssl_dir / "server.crt"), moved)
|
|
self.assertIn((self.paths.legacy_key,
|
|
self.paths.ssl_dir / "server.key"), moved)
|
|
self.assertEqual(
|
|
(self.paths.ssl_dir / "server.key").stat().st_mode & 0o777,
|
|
0o640)
|
|
|
|
def test_migration_rejects_incomplete_legacy_tls_pair(self):
|
|
self.paths.legacy_key.parent.mkdir(parents=True)
|
|
self.paths.legacy_key.write_text("private key")
|
|
with self.assertRaisesRegex(ConfigurationError, "requires both"):
|
|
self.store.migrate_legacy()
|
|
|
|
def test_alias_fragments_allow_either_section_or_both(self):
|
|
self.paths.aliases_dir.mkdir(parents=True)
|
|
(self.paths.aliases_dir / "default_config").write_bytes(json_bytes({
|
|
"aliases": {"postmaster": "admin"},
|
|
}))
|
|
(self.paths.aliases_dir / "domains").write_bytes(json_bytes({
|
|
"domains": {"old.example.test": "example.test"},
|
|
}))
|
|
(self.paths.aliases_dir / "special.example.test").write_bytes(
|
|
json_bytes({
|
|
"aliases": {"dmarc-reports@example.test": "dmarc"},
|
|
"domains": {
|
|
"special.example.test": {
|
|
"target": "example.test",
|
|
"recipient_mapping": "full-address",
|
|
},
|
|
},
|
|
}))
|
|
aliases = self.store.load_aliases()
|
|
self.assertEqual(
|
|
aliases["special.example.test"]["domains"]
|
|
["special.example.test"]["recipient_mapping"],
|
|
"full-address")
|
|
|
|
def test_alias_domains_compare_by_idna2008_alabel(self):
|
|
self.paths.aliases_dir.mkdir(parents=True)
|
|
(self.paths.aliases_dir / "default_config").write_bytes(json_bytes({
|
|
"aliases": {"postmaster@bücher.example": "admin"},
|
|
"domains": {"bücher.example": "xn--bcher-kva.example"},
|
|
}))
|
|
aliases = self.store.load_aliases(["xn--bcher-kva.example"])
|
|
self.assertIn("postmaster@bücher.example",
|
|
aliases["default_config"]["aliases"])
|
|
|
|
def test_alias_fragments_reject_numeric_recipient_mapping(self):
|
|
with self.assertRaisesRegex(ConfigurationError, "invalid options"):
|
|
self.store._validate_alias_document("legacy", {
|
|
"domains": {
|
|
"old.example.test": {
|
|
"target": "example.test",
|
|
"recipient_mapping": 1,
|
|
},
|
|
},
|
|
})
|
|
|
|
def test_alias_fragments_reject_cross_file_conflicts(self):
|
|
self.paths.aliases_dir.mkdir(parents=True)
|
|
(self.paths.aliases_dir / "default_config").write_bytes(json_bytes({
|
|
"aliases": {"dmarc-reports": "admin"},
|
|
}))
|
|
(self.paths.aliases_dir / "conflict").write_bytes(json_bytes({
|
|
"aliases": {"dmarc-reports": "dmarc"},
|
|
}))
|
|
with self.assertRaisesRegex(ConfigurationError, "conflicting alias"):
|
|
self.store.load_aliases(["example.test"])
|
|
|
|
def test_alias_fragments_reject_unhosted_domain_override(self):
|
|
self.paths.aliases_dir.mkdir(parents=True)
|
|
(self.paths.aliases_dir / "default_config").write_bytes(json_bytes({
|
|
"aliases": {"postmaster": "admin"},
|
|
}))
|
|
(self.paths.aliases_dir / "override").write_bytes(json_bytes({
|
|
"aliases": {"dmarc-reports@outside.test": "dmarc"},
|
|
}))
|
|
with self.assertRaisesRegex(ConfigurationError, "unhosted domain"):
|
|
self.store.load_aliases(["example.test"])
|
|
|
|
def test_dmarc_tasks_follow_domain_specific_alias_targets(self):
|
|
self.paths.aliases_dir.mkdir(parents=True)
|
|
(self.paths.aliases_dir / "default_config").write_bytes(json_bytes({
|
|
"aliases": {"dmarc-reports": "admin"},
|
|
}))
|
|
(self.paths.aliases_dir / "reports").write_bytes(json_bytes({
|
|
"aliases": {
|
|
"dmarc-reports@second.example.test":
|
|
"reports@second.example.test",
|
|
},
|
|
}))
|
|
self.assertEqual(self.store.dmarc_report_users([
|
|
"example.test", "second.example.test",
|
|
]), {
|
|
"example.test": "admin",
|
|
"second.example.test": "reports",
|
|
})
|
|
|
|
def test_migration_names_legacy_recipient_mapping(self):
|
|
self.store.username_mappings["example.test"] = 2
|
|
self.store.migrate_legacy()
|
|
migrated = json.loads(
|
|
(self.paths.aliases_dir / "example.test").read_text())
|
|
self.assertEqual(
|
|
migrated["domains"]["example.test"]["recipient_mapping"],
|
|
"domain-name")
|
|
|
|
def test_migration_rejects_unknown_legacy_recipient_mapping(self):
|
|
self.store.username_mappings["example.test"] = 3
|
|
with self.assertRaisesRegex(ConfigurationError, "invalid structure"):
|
|
self.store.migrate_legacy()
|
|
|
|
def test_retained_alias_transaction_commits_persistent_mapping(self):
|
|
self.paths.aliases_dir.mkdir(parents=True)
|
|
(self.paths.aliases_dir / "default_config").write_bytes(json_bytes({
|
|
"aliases": {"postmaster": "admin"},
|
|
}))
|
|
self.store.prepare_retained_alias("old-user", "new-user", "101")
|
|
aliases = json.loads(
|
|
(self.paths.aliases_dir / "account-renames").read_text())
|
|
self.assertEqual(aliases["aliases"]["old-user"], "new-user")
|
|
self.store.finish_retained_alias("101", True)
|
|
self.assertFalse(
|
|
(self.paths.lock.parent / "bongo-alias-rename-101").exists())
|
|
self.assertEqual(json.loads(
|
|
(self.paths.aliases_dir / "account-renames").read_text()
|
|
)["aliases"]["old-user"], "new-user")
|
|
|
|
def test_retained_alias_transaction_rolls_back_exact_file(self):
|
|
self.paths.aliases_dir.mkdir(parents=True)
|
|
(self.paths.aliases_dir / "default_config").write_bytes(json_bytes({
|
|
"aliases": {"postmaster": "admin"},
|
|
}))
|
|
self.store.prepare_retained_alias("old-user", "new-user", "102")
|
|
self.store.finish_retained_alias("102", False)
|
|
self.assertFalse((self.paths.aliases_dir / "account-renames").exists())
|
|
|
|
def test_retained_alias_prepare_rejects_conflicting_mapping(self):
|
|
self.paths.aliases_dir.mkdir(parents=True)
|
|
(self.paths.aliases_dir / "default_config").write_bytes(json_bytes({
|
|
"aliases": {"old-user": "someone-else"},
|
|
}))
|
|
with self.assertRaisesRegex(ConfigurationError, "already points"):
|
|
self.store.prepare_retained_alias("old-user", "new-user", "103")
|
|
|
|
def test_retained_alias_rejects_control_characters(self):
|
|
with self.assertRaisesRegex(ConfigurationError, "invalid retained"):
|
|
self.store.prepare_retained_alias(
|
|
"old\x7fuser", "new-user", "104")
|
|
|
|
def test_store_failure_rolls_back_earlier_documents(self):
|
|
configurations = copy.deepcopy(self.store.live)
|
|
original_smtp = copy.deepcopy(self.store.live["smtp"])
|
|
original_imap = copy.deepcopy(self.store.live["imap"])
|
|
configurations["smtp"]["socket_timeout"] += 1
|
|
configurations["imap"]["threads_max"] += 1
|
|
self.store.fail_name = "imap"
|
|
with self.assertRaises(ConfigurationError):
|
|
self.store.save(configurations)
|
|
self.assertEqual(self.store.live["smtp"], original_smtp)
|
|
self.assertEqual(self.store.live["imap"], original_imap)
|
|
self.assertTrue(all(
|
|
not self.paths.document(name).exists()
|
|
for name in STORE_DOCUMENTS
|
|
))
|
|
|
|
def test_store_helper_timeout_names_operation_and_document(self):
|
|
store = ConfigurationStore(self.paths)
|
|
with mock.patch("subprocess.run", side_effect=subprocess.TimeoutExpired(
|
|
[str(self.paths.admin), "__config-read", "sieve"], 30)):
|
|
with self.assertRaisesRegex(
|
|
ConfigurationError,
|
|
r"timed out during __config-read for sieve"):
|
|
store.read_store_document("sieve")
|
|
|
|
def test_scanner_ping_probes_require_expected_reply(self):
|
|
clam = FakeSocket(b"PONG\0")
|
|
spam = FakeSocket(b"SPAMD/1.5 0 PONG\r\n")
|
|
with mock.patch("socket.create_connection",
|
|
side_effect=[clam, spam]):
|
|
self.assertTrue(probe_clamav().available)
|
|
self.assertTrue(probe_spamd().available)
|
|
self.assertEqual(clam.sent, b"zPING\0")
|
|
self.assertEqual(spam.sent, b"PING SPAMC/1.5\r\n\r\n")
|
|
|
|
def test_scanner_probe_fails_closed_on_protocol_mismatch(self):
|
|
with mock.patch("socket.create_connection",
|
|
return_value=FakeSocket(b"unexpected\n")), \
|
|
mock.patch("shutil.which", return_value="/usr/bin/clamd"):
|
|
result = probe_clamav()
|
|
self.assertFalse(result.available)
|
|
self.assertTrue(result.installed)
|
|
|
|
def test_scanner_probe_reports_installed_but_unreachable(self):
|
|
with mock.patch("socket.create_connection",
|
|
side_effect=ConnectionRefusedError("refused")), \
|
|
mock.patch("shutil.which", return_value="/usr/bin/spamd"):
|
|
result = probe_spamd()
|
|
self.assertFalse(result.available)
|
|
self.assertTrue(result.installed)
|
|
self.assertEqual(result.endpoint, "127.0.0.1:783")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|