diff --git a/src/apps/config/config/web b/src/apps/config/config/web index a2d3aeb..01eba36 100644 --- a/src/apps/config/config/web +++ b/src/apps/config/config/web @@ -6,6 +6,7 @@ "http_redirect_port": 80, "public_base_url": "", "external_https": false, + "login_domains": [], "allow_basic_auth": true, "trusted_proxies": ["127.0.0.1", "::1"], "client_ip_header": "X-Forwarded-For", diff --git a/src/apps/config/tests/test_configuration.py b/src/apps/config/tests/test_configuration.py index 1a71051..f3daf80 100644 --- a/src/apps/config/tests/test_configuration.py +++ b/src/apps/config/tests/test_configuration.py @@ -35,6 +35,7 @@ from bongo.configuration.model import ( # noqa: E402 DOCUMENT_ORDER, Scenario, apply_scenario, + canonicalize_domains, enable_scanner_agents, errors, load_templates, @@ -106,11 +107,37 @@ class ConfigurationModelTest(unittest.TestCase): ["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_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( diff --git a/src/libs/python/bongo/configuration/model.py b/src/libs/python/bongo/configuration/model.py index f870227..b332b38 100644 --- a/src/libs/python/bongo/configuration/model.py +++ b/src/libs/python/bongo/configuration/model.py @@ -223,6 +223,12 @@ def canonicalize_domains(configurations: Mapping[str, Any]) -> \ for key in ("domains", "hosteddomains"): if isinstance(queue.get(key), list): queue[key] = [domain(value) for value in queue[key]] + hosted_domains = queue.get("hosteddomains") + if isinstance(hosted_domains, list): + # This is a runtime mirror for the Web agent, not a second source of + # truth. Keeping it here also upgrades pre-0.7 configuration during + # the next `bongo-admin config sync`. + result.get("web", {})["login_domains"] = copy.deepcopy(hosted_domains) smtp = result.get("smtp", {}) for key in ("internal_relay_domain", "dkim_signing_domain", "srs_domain"): @@ -323,6 +329,7 @@ def apply_scenario(templates: Mapping[str, Any], scenario: Scenario) -> dict[str job["enabled"] = scenario.acme_enabled web = configs["web"] + web["login_domains"] = domains if scenario.web_mode in {"direct", "direct_https"}: web["listen_address"] = scenario.bind_address web["trusted_proxies"] = [] @@ -867,6 +874,8 @@ def _validate_web(config: Mapping[str, Any], issues: list[Issue], if not _valid_ipv4(config.get("listen_address")): _issue(issues, "web", "listen_address", "Bongo 0.7 requires an IPv4 address") _port(issues, "web", config, "port") + _validate_domains(issues, "web", "login_domains", + config.get("login_domains")) _validate_networks(issues, "web", "trusted_proxies", config.get("trusted_proxies"), False) header = config.get("client_ip_header") if not isinstance(header, str) or not _HEADER_NAME.fullmatch(header): diff --git a/src/www/bongo-web.py.in b/src/www/bongo-web.py.in index 0cf7c3f..967f61e 100644 --- a/src/www/bongo-web.py.in +++ b/src/www/bongo-web.py.in @@ -75,6 +75,7 @@ from bongo_web.external_accounts import (create_account as create_external_accou list_identities as list_external_identities) from bongo_web.security import (SECOND_FACTOR_VALID, ExpiringSecrets, disable_totp, enable_totp, + hosted_login_username, legacy_login_clients, new_totp_enrollment, policy, resolve_user_alias, verify_second_factor) @@ -750,14 +751,16 @@ class BongoWebHandler(BaseHTTPRequestHandler): token, lifetime, self._secure_cookies()), "Vary": "Cookie"}) return - user = str(data.get("user", "")).strip() - requested_user = user + requested_user = str(data.get("user", "")).strip() + authentication_user = hosted_login_username( + requested_user, self.server.config.get("login_domains", [])) + user = authentication_user try: user = resolve_user_alias(user) except (RuntimeError, ValueError): - user = requested_user + user = authentication_user password = str(data.get("password", "")) - key = (source_address, requested_user.casefold()) + key = (source_address, authentication_user.casefold()) if not self.server.login_limiter.allowed(key): self._json(HTTPStatus.TOO_MANY_REQUESTS, {"error": "authentication temporarily unavailable"}, @@ -774,7 +777,8 @@ class BongoWebHandler(BaseHTTPRequestHandler): return cookie_lifetime = min(lifetime, int(self.server.config.get( "second_factor_lifetime", 300))) if second_factor_required else lifetime - store_cookie = authenticate(requested_user, password, cookie_lifetime) + store_cookie = authenticate(authentication_user, password, + cookie_lifetime) password = None if store_cookie is None: self.server.login_limiter.failed(key) diff --git a/src/www/bongo_web/security.py b/src/www/bongo_web/security.py index 7cf6f91..fc44379 100644 --- a/src/www/bongo_web/security.py +++ b/src/www/bongo_web/security.py @@ -26,6 +26,7 @@ import threading import time from urllib.parse import quote, urlencode +from bongo.domain import domain_to_ascii from libbongo import _auth_security, _qrcode @@ -101,6 +102,25 @@ def resolve_user_alias(username): return _auth_security.resolve_user_alias(username) +def hosted_login_username(username, domains): + """Map a mailbox in a configured hosted domain to its Store username.""" + username = str(username).strip() + if username.count("@") != 1: + return username + local, domain = username.rsplit("@", 1) + if not local or not domain: + return username + try: + requested_domain = domain_to_ascii(domain) + hosted_domains = { + domain_to_ascii(value) for value in domains + if isinstance(value, str) + } + except ValueError: + return username + return local if requested_domain in hosted_domains else username + + def legacy_login_clients(username): return _auth_security.list_legacy_login_clients(username) diff --git a/src/www/tests/test_security.py b/src/www/tests/test_security.py index e359b67..04b0975 100644 --- a/src/www/tests/test_security.py +++ b/src/www/tests/test_security.py @@ -89,6 +89,31 @@ class EnrollmentTest(unittest.TestCase): security.new_totp_enrollment("mario@example.invalid", "bad\nissuer") +class HostedLoginUsernameTest(unittest.TestCase): + def test_local_username_is_unchanged(self): + self.assertEqual( + security.hosted_login_username("admin", ["example.test"]), + "admin") + + def test_hosted_mailbox_maps_to_local_username(self): + self.assertEqual(security.hosted_login_username( + "admin@Example.Test", ["example.test"]), "admin") + + def test_idna_domain_maps_to_local_username(self): + self.assertEqual(security.hosted_login_username( + "mario@bücher.example", ["xn--bcher-kva.example"]), "mario") + + def test_unknown_domain_is_not_stripped(self): + self.assertEqual(security.hosted_login_username( + "admin@attacker.example", ["example.test"]), + "admin@attacker.example") + + def test_malformed_mailbox_is_not_rewritten(self): + self.assertEqual(security.hosted_login_username( + "admin@@example.test", ["example.test"]), + "admin@@example.test") + + class LoginMarkupTest(unittest.TestCase): def test_second_factor_field_is_hidden_initially(self): markup = (Path(__file__).parents[1] / "static" / "index.html").read_text(