351 lines
16 KiB
Python
351 lines
16 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>
|
|
# ****************************************************************************/
|
|
|
|
"""Tests for the shared setup/admin configuration model."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import sys
|
|
import unittest
|
|
from unittest import mock
|
|
from pathlib import Path
|
|
|
|
SOURCE = Path(__file__).resolve().parents[4]
|
|
sys.path.insert(0, str(SOURCE / "src" / "libs" / "python"))
|
|
|
|
from bongo.configuration.model import ( # noqa: E402
|
|
DOCUMENT_ORDER,
|
|
Scenario,
|
|
apply_scenario,
|
|
canonicalize_domains,
|
|
enable_scanner_agents,
|
|
errors,
|
|
load_templates,
|
|
parse_legacy_authentication,
|
|
validate_all,
|
|
)
|
|
from bongo.domain import domain_to_ascii # noqa: E402
|
|
|
|
|
|
def templates():
|
|
root = SOURCE / "src" / "apps" / "config" / "config"
|
|
return load_templates(root)
|
|
|
|
|
|
class ConfigurationModelTest(unittest.TestCase):
|
|
def test_idna2008_uses_nontransitional_sharp_s(self):
|
|
self.assertEqual(domain_to_ascii("faß.de"), "xn--fa-hia.de")
|
|
|
|
def test_all_template_names_are_suffixless(self):
|
|
root = SOURCE / "src" / "apps" / "config" / "config"
|
|
self.assertEqual(set(DOCUMENT_ORDER), {
|
|
path.name for path in root.iterdir() if path.is_file()
|
|
})
|
|
|
|
def test_safe_reverse_proxy_scenario(self):
|
|
defaults = templates()
|
|
scenario = Scenario(
|
|
hostname="mail.example.test",
|
|
bind_address="172.16.11.14",
|
|
domains=["example.test"],
|
|
admin_password="correct horse battery staple",
|
|
web_mode="apache",
|
|
web_public_url="https://mail.example.test",
|
|
web_proxy_networks=["127.0.0.1/32"],
|
|
mail_proxy_enabled=True,
|
|
mail_proxy_networks=["172.16.11.7/32"],
|
|
internal_relay_enabled=True,
|
|
internal_relay_networks=["172.16.11.0/24"],
|
|
internal_relay_domain="example.test",
|
|
antispam_enabled=True,
|
|
antivirus_enabled=True,
|
|
)
|
|
configs = apply_scenario(defaults, scenario)
|
|
self.assertEqual(configs["smtp"]["proxy_protocol_networks"],
|
|
["172.16.11.7/32"])
|
|
self.assertEqual(configs["global"]["tls_hostnames"],
|
|
["mail.example.test"])
|
|
self.assertEqual(configs["web"]["listen_address"], "127.0.0.1")
|
|
self.assertTrue(configs["antispam"]["enabled"])
|
|
self.assertFalse(errors(validate_all(configs, defaults, check_files=False)))
|
|
|
|
def test_international_domains_are_stored_as_idna2008_alabels(self):
|
|
defaults = templates()
|
|
configs = apply_scenario(defaults, Scenario(
|
|
hostname="mail.bücher.example",
|
|
bind_address="192.0.2.10",
|
|
domains=["bücher.example", "xn--caf-dma.example"],
|
|
tls_hostnames=["mail.bücher.example"],
|
|
admin_password="0123456789ab",
|
|
internal_relay_enabled=True,
|
|
internal_relay_networks=["192.0.2.0/24"],
|
|
internal_relay_domain="bücher.example",
|
|
web_mode="apache",
|
|
web_public_url="https://mail.xn--bcher-kva.example",
|
|
))
|
|
self.assertEqual(configs["global"]["hostname"],
|
|
"mail.xn--bcher-kva.example")
|
|
self.assertEqual(configs["global"]["tls_hostnames"],
|
|
["mail.xn--bcher-kva.example"])
|
|
self.assertEqual(configs["queue"]["domains"], [
|
|
"xn--bcher-kva.example", "xn--caf-dma.example"])
|
|
self.assertEqual(configs["web"]["login_domains"], [
|
|
"xn--bcher-kva.example", "xn--caf-dma.example"])
|
|
self.assertEqual(configs["smtp"]["internal_relay_domain"],
|
|
"xn--bcher-kva.example")
|
|
self.assertFalse(errors(validate_all(
|
|
configs, defaults, check_files=False)))
|
|
|
|
def test_invalid_web_login_domain_is_rejected(self):
|
|
defaults = templates()
|
|
configs = apply_scenario(defaults, 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",
|
|
))
|
|
configs["web"]["login_domains"] = ["not a domain"]
|
|
messages = [str(issue) for issue in validate_all(
|
|
configs, defaults, check_files=False)]
|
|
self.assertTrue(any("web.login_domains[0]" in message
|
|
for message in messages))
|
|
|
|
def test_web_administrators_are_explicit_and_validated(self):
|
|
defaults = templates()
|
|
self.assertEqual(defaults["web"]["administrators"], ["admin"])
|
|
configs = apply_scenario(defaults, Scenario(
|
|
hostname="mail.example.test", bind_address="192.0.2.10",
|
|
domains=["example.test"], admin_password="0123456789ab",
|
|
))
|
|
configs["web"]["administrators"] = []
|
|
messages = [str(issue) for issue in validate_all(
|
|
configs, defaults, check_files=False)]
|
|
self.assertTrue(any("web.administrators" in message
|
|
for message in messages))
|
|
|
|
def test_web_login_domains_follow_hosted_mail_domains_on_sync(self):
|
|
configs = apply_scenario(templates(), Scenario(
|
|
hostname="mail.example.test", bind_address="192.0.2.10",
|
|
domains=["example.test"], admin_password="0123456789ab",
|
|
))
|
|
configs["queue"]["hosteddomains"] = ["bücher.example"]
|
|
configs["web"]["login_domains"] = []
|
|
synchronized = canonicalize_domains(configs)
|
|
self.assertEqual(synchronized["web"]["login_domains"],
|
|
["xn--bcher-kva.example"])
|
|
|
|
def test_verified_scanners_enable_bongo_agents_and_local_endpoints(self):
|
|
defaults = templates()
|
|
configs = apply_scenario(defaults, 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",
|
|
))
|
|
enable_scanner_agents(configs, {"ClamAV", "SpamAssassin"})
|
|
agents = {
|
|
agent["name"]: agent["enabled"]
|
|
for agent in configs["manager"]["agents"]
|
|
}
|
|
self.assertTrue(agents["bongoantispam"])
|
|
self.assertTrue(agents["bongoavirus"])
|
|
self.assertTrue(configs["antispam"]["enabled"])
|
|
self.assertEqual(configs["antispam"]["hosts"], ["127.0.0.1"])
|
|
self.assertEqual(configs["antivirus"]["hosts"],
|
|
["127.0.0.1:3310:1"])
|
|
self.assertFalse(errors(validate_all(
|
|
configs, defaults, check_files=False)))
|
|
|
|
def test_web_databases_follow_configured_state_directory(self):
|
|
defaults = templates()
|
|
with mock.patch.dict("os.environ", {
|
|
"BONGO_STATE_DIR": "/srv/bongo-test-state"}):
|
|
configs = apply_scenario(defaults, Scenario(
|
|
hostname="mail.example.test",
|
|
bind_address="127.0.0.1",
|
|
domains=["example.test"],
|
|
admin_password="0123456789ab",
|
|
web_mode="direct",
|
|
web_public_url="http://127.0.0.1:8080",
|
|
))
|
|
self.assertEqual(
|
|
configs["web"]["session_database"],
|
|
"/srv/bongo-test-state/web/sessions.db")
|
|
self.assertEqual(
|
|
configs["web"]["task_database"],
|
|
"/srv/bongo-test-state/web/tasks.db")
|
|
self.assertEqual(
|
|
configs["web"]["identity_database"],
|
|
"/srv/bongo-test-state/web/identities.db")
|
|
|
|
def test_proxy_requires_trusted_network(self):
|
|
defaults = templates()
|
|
configs = apply_scenario(defaults, 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",
|
|
mail_proxy_enabled=True,
|
|
))
|
|
messages = [str(issue) for issue in validate_all(configs, defaults, False)]
|
|
self.assertTrue(any("proxy_protocol_networks" in message for message in messages))
|
|
|
|
def test_mta_sts_cache_requires_absolute_path(self):
|
|
defaults = templates()
|
|
configs = apply_scenario(defaults, 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",
|
|
))
|
|
configs["smtp"]["outbound_mta_sts_cache_directory"] = "relative/cache"
|
|
messages = [str(issue) for issue in validate_all(configs, defaults, False)]
|
|
self.assertTrue(any("outbound_mta_sts_cache_directory" in message
|
|
for message in messages))
|
|
|
|
def test_tls_reporting_requires_registered_job_and_safe_organization(self):
|
|
defaults = templates()
|
|
configs = apply_scenario(defaults, 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",
|
|
))
|
|
configs["worker"]["tls_reporting"]["organization_name"] = "bad\nname"
|
|
configs["worker"]["jobs"] = [
|
|
job for job in configs["worker"]["jobs"]
|
|
if job["name"] != "tlsrpt-delivery"
|
|
]
|
|
messages = [str(issue) for issue in
|
|
validate_all(configs, defaults, False)]
|
|
self.assertTrue(any("organization_name" in message
|
|
for message in messages))
|
|
self.assertTrue(any("tlsrpt-delivery" in message
|
|
for message in messages))
|
|
|
|
def test_direct_https_binds_443_and_redirects_80(self):
|
|
defaults = templates()
|
|
configs = apply_scenario(defaults, Scenario(
|
|
hostname="mail.example.test", bind_address="192.0.2.10",
|
|
domains=["example.test"], admin_password="0123456789ab",
|
|
web_mode="direct_https",
|
|
web_public_url="https://mail.example.test",
|
|
web_tls_certificate="/etc/bongo/tls/fullchain.pem",
|
|
web_tls_key="/etc/bongo/tls/private-key.pem",
|
|
))
|
|
web = configs["web"]
|
|
self.assertEqual(web["port"], 443)
|
|
self.assertEqual(web["http_redirect_port"], 80)
|
|
self.assertEqual(web["listen_address"], "192.0.2.10")
|
|
self.assertTrue(web["external_https"])
|
|
self.assertFalse(errors(validate_all(configs, defaults,
|
|
check_files=False)))
|
|
|
|
def test_listener_port_collision_is_rejected(self):
|
|
defaults = templates()
|
|
configs = apply_scenario(defaults, 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",
|
|
))
|
|
configs["imap"]["port"] = configs["smtp"]["port"]
|
|
messages = [str(issue) for issue in validate_all(configs, defaults, False)]
|
|
self.assertTrue(any("listener port 25" in message for message in messages))
|
|
|
|
def test_ldap_autoprovision_requires_group(self):
|
|
defaults = templates()
|
|
configs = apply_scenario(defaults, Scenario(
|
|
hostname="mail.example.test", bind_address="192.0.2.10",
|
|
domains=["example.test"], admin_password="0123456789ab",
|
|
ldap_enabled=True, ldap_uri="ldaps://ldap.example.test",
|
|
ldap_base_dn="dc=example,dc=test", ldap_auto_provision=True,
|
|
web_mode="direct", web_public_url="http://mail.example.test:8080",
|
|
))
|
|
messages = [str(issue) for issue in validate_all(configs, defaults, False)]
|
|
self.assertTrue(any("AuthLdapRequiredGroupDn" in message for message in messages))
|
|
|
|
def test_legacy_authentication_parser_is_migration_only(self):
|
|
defaults = templates()["authentication"]
|
|
original = ("Unrelated = yes\nAuthLdapEnabled = true\n"
|
|
"AuthLdapOfflineCacheHours = 72\n")
|
|
parsed = parse_legacy_authentication(original, defaults)
|
|
self.assertTrue(parsed["AuthLdapEnabled"])
|
|
self.assertEqual(parsed["AuthLdapOfflineCacheHours"], 72)
|
|
self.assertNotIn("Unrelated", parsed)
|
|
|
|
def test_setup_enables_delivery_rules_independently_of_managesieve(self):
|
|
scenario = Scenario(
|
|
hostname="mail.example.test", bind_address="192.0.2.5",
|
|
domains=["example.test"], admin_password="0123456789ab",
|
|
sieve_enabled=False,
|
|
)
|
|
configs = apply_scenario(templates(), scenario)
|
|
agents = {
|
|
agent["name"]: agent["enabled"]
|
|
for agent in configs["manager"]["agents"]
|
|
}
|
|
self.assertTrue(agents["bongorules"])
|
|
self.assertFalse(agents["bongosieve"])
|
|
|
|
def test_per_sender_domain_dkim_signing_is_valid(self):
|
|
defaults = templates()
|
|
scenario = Scenario(
|
|
hostname="mail.example.test", bind_address="192.0.2.5",
|
|
domains=["example.test"], admin_password="0123456789ab",
|
|
dkim_selector="bongo",
|
|
dkim_key_directory="/etc/bongo/dkim.d",
|
|
web_mode="direct",
|
|
web_public_url="http://mail.example.test:8080",
|
|
)
|
|
configs = apply_scenario(defaults, scenario)
|
|
self.assertTrue(configs["smtp"]["dkim_sign_outgoing"])
|
|
self.assertEqual(configs["smtp"]["dkim_signing_domain"], "")
|
|
self.assertFalse(errors(validate_all(
|
|
configs, defaults, check_files=False)))
|
|
|
|
def test_dns_acme_scenario_selects_dns_challenge(self):
|
|
defaults = templates()
|
|
scenario = Scenario(
|
|
hostname="mail.example.test", bind_address="192.0.2.5",
|
|
domains=["example.test"], admin_password="0123456789ab",
|
|
acme_challenge_type="dns-01",
|
|
web_mode="direct",
|
|
web_public_url="http://mail.example.test:8080",
|
|
)
|
|
configs = apply_scenario(defaults, scenario)
|
|
self.assertEqual(configs["acme"]["challenge_type"], "dns-01")
|
|
self.assertFalse(errors(validate_all(
|
|
configs, defaults, check_files=False)))
|
|
|
|
def test_acme_rejects_unknown_challenge_and_bad_dns_polling(self):
|
|
defaults = templates()
|
|
configs = apply_scenario(defaults, Scenario(
|
|
hostname="mail.example.test", bind_address="192.0.2.5",
|
|
domains=["example.test"], admin_password="0123456789ab",
|
|
))
|
|
configs["acme"]["challenge_type"] = "dns-provider-command"
|
|
configs["acme"]["dns_poll_interval_seconds"] = 1201
|
|
messages = [str(issue) for issue in validate_all(
|
|
configs, defaults, check_files=False)]
|
|
self.assertTrue(any("challenge_type" in message
|
|
for message in messages))
|
|
self.assertTrue(any("dns_poll_interval_seconds" in message
|
|
for message in messages))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|