Accept hosted email addresses for web login

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",

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(

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):

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)

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)
 

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(