geos_one/bongo
1
0
Fork 0
Code Issues Pull Requests Actions 7 Packages Projects Releases 1 Wiki Activity
Files
92e0ba117d393e67906a6d632debbb4fa3b5ca54
bongo/src/apps/config/tests/test_configuration.py
T
Mario Fetka 92e0ba117d
Debian Trixie package bundle / packages (push) Failing after 14m43s
Details
Bound external collector queue growth
2026-07-23 22:15:21 +02:00

480 lines
22 KiB
Python
Raw Blame History

# /****************************************************************************
# * <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 re
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_native_admin_allows_every_configuration_document(self):
source = (SOURCE / "src" / "apps" / "admin" / "admin.c").read_text(
encoding="utf-8")
match = re.search(
r"static const char \*const allowed\[\] = \{(.*?)NULL",
source, re.DOTALL)
self.assertIsNotNone(match)
allowed = set(re.findall(r'"([a-z0-9-]+)"', match.group(1)))
self.assertEqual(set(DOCUMENT_ORDER), allowed)
def test_manager_reconciles_every_configuration_document(self):
source = (SOURCE / "src" / "apps" / "manager" / "manager.c").read_text(
encoding="utf-8")
match = re.search(
r"SyncLocalConfiguration\(\).*?"
r"static const char \*const names\[\] = \{(.*?)NULL",
source, re.DOTALL)
self.assertIsNotNone(match)
reconciled = set(re.findall(r'"([a-z0-9-]+)"', match.group(1)))
self.assertEqual(set(DOCUMENT_ORDER), reconciled)
def test_collector_private_ca_requires_absolute_path(self):
defaults = templates()
configs = templates()
configs["collector"]["ca_file"] = "relative-ca.pem"
problems = errors(validate_all(configs, defaults, check_files=False))
self.assertTrue(any(issue.document == "collector" and
issue.path == "ca_file" for issue in problems))
configs["collector"]["ca_file"] = "/etc/ssl/private-provider-ca.pem"
self.assertFalse(any(issue.document == "collector" for issue in
errors(validate_all(configs, defaults,
check_files=False))))
def test_collector_pending_limits_are_positive_and_ordered(self):
defaults = templates()
configs = templates()
configs["collector"]["maximum_pending_messages"] = 10
configs["collector"]["maximum_pending_messages_per_user"] = 11
problems = errors(validate_all(configs, defaults, check_files=False))
self.assertTrue(any(
issue.document == "collector" and
issue.path == "maximum_pending_messages_per_user"
for issue in problems))
configs["collector"]["maximum_pending_messages_per_user"] = 5
self.assertFalse(any(
issue.document == "collector" and
issue.path.startswith("maximum_pending_messages")
for issue in errors(validate_all(
configs, defaults, check_files=False))))
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_all_web_reverse_proxy_modes_use_the_same_safe_boundary(self):
defaults = templates()
for mode in ("apache", "nginx", "npm"):
with self.subTest(mode=mode):
configs = apply_scenario(defaults, Scenario(
hostname="mail.example.test",
bind_address="192.0.2.10",
domains=["example.test"],
admin_password="0123456789ab",
web_mode=mode,
web_public_url="https://mail.example.test",
web_proxy_networks=["10.20.30.40/32"],
))
web = configs["web"]
self.assertEqual(web["listen_address"], "127.0.0.1")
self.assertEqual(web["port"], 8080)
self.assertEqual(web["trusted_proxies"], ["10.20.30.40/32"])
self.assertEqual(web["client_ip_header"], "X-Forwarded-For")
self.assertTrue(web["external_https"])
self.assertEqual(web["tls_certificate"], "")
self.assertEqual(web["tls_key"], "")
self.assertFalse(errors(validate_all(
configs, defaults, check_files=False)))
def test_haproxy_boundary_is_applied_to_every_mail_listener(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_public_url="https://mail.example.test",
mail_proxy_enabled=True,
mail_proxy_networks=["10.20.30.40/32"],
))
for name in ("smtp", "imap", "pop3", "sieve"):
with self.subTest(agent=name):
self.assertTrue(configs[name]["proxy_protocol_enabled"])
self.assertEqual(configs[name]["proxy_protocol_networks"],
["10.20.30.40/32"])
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_direct_http_is_explicitly_warned_but_not_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",
))
web = configs["web"]
self.assertEqual(web["listen_address"], "192.0.2.10")
self.assertEqual(web["port"], 8080)
self.assertEqual(web["trusted_proxies"], [])
self.assertFalse(web["external_https"])
issues = validate_all(configs, defaults, check_files=False)
self.assertFalse(errors(issues))
self.assertTrue(any(issue.document == "web" and
issue.path == "external_https" and
issue.severity == "warning"
for issue in issues))
def test_direct_https_rejects_incomplete_key_pair(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/ssl.d/server.crt",
web_tls_key="",
))
messages = [str(issue) for issue in validate_all(
configs, defaults, check_files=False)]
self.assertTrue(any("direct HTTPS requires both" in message
for message in messages))
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()
Reference in New Issue View Git Blame Copy Permalink