Fix domain mapping bugs and improve UI/UX: normalize domains, fix delete, add validation and search

app/admin_api.py

@@ -17,7 +17,7 @@ from .extensions import limiter
 from .iam import IamError, Principal
 from .replication import ReplicationManager
 from .site_registry import PeerSite, SiteInfo, SiteRegistry
-from .website_domains import WebsiteDomainStore
+from .website_domains import WebsiteDomainStore, normalize_domain, is_valid_domain
 
 
 def _is_safe_url(url: str, allow_internal: bool = False) -> bool:
@@ -704,10 +704,12 @@ def create_website_domain():
     if not current_app.config.get("WEBSITE_HOSTING_ENABLED", False):
         return _json_error("InvalidRequest", "Website hosting is not enabled", 400)
     payload = request.get_json(silent=True) or {}
-    domain = (payload.get("domain") or "").strip().lower()
+    domain = normalize_domain(payload.get("domain") or "")
     bucket = (payload.get("bucket") or "").strip()
     if not domain:
         return _json_error("ValidationError", "domain is required", 400)
+    if not is_valid_domain(domain):
+        return _json_error("ValidationError", f"Invalid domain: '{domain}'", 400)
     if not bucket:
         return _json_error("ValidationError", "bucket is required", 400)
     storage = _storage()
@@ -730,10 +732,11 @@ def get_website_domain(domain: str):
         return error
     if not current_app.config.get("WEBSITE_HOSTING_ENABLED", False):
         return _json_error("InvalidRequest", "Website hosting is not enabled", 400)
+    domain = normalize_domain(domain)
     bucket = _website_domains().get_bucket(domain)
     if not bucket:
         return _json_error("NotFound", f"No mapping found for domain '{domain}'", 404)
-    return jsonify({"domain": domain.lower(), "bucket": bucket})
+    return jsonify({"domain": domain, "bucket": bucket})
 
 
 @admin_api_bp.route("/website-domains/<domain>", methods=["PUT"])
@@ -744,6 +747,7 @@ def update_website_domain(domain: str):
         return error
     if not current_app.config.get("WEBSITE_HOSTING_ENABLED", False):
         return _json_error("InvalidRequest", "Website hosting is not enabled", 400)
+    domain = normalize_domain(domain)
     payload = request.get_json(silent=True) or {}
     bucket = (payload.get("bucket") or "").strip()
     if not bucket:
@@ -752,9 +756,11 @@ def update_website_domain(domain: str):
     if not storage.bucket_exists(bucket):
         return _json_error("NoSuchBucket", f"Bucket '{bucket}' does not exist", 404)
     store = _website_domains()
+    if not store.get_bucket(domain):
+        return _json_error("NotFound", f"No mapping found for domain '{domain}'", 404)
     store.set_mapping(domain, bucket)
     logger.info("Website domain mapping updated: %s -> %s", domain, bucket)
-    return jsonify({"domain": domain.lower(), "bucket": bucket})
+    return jsonify({"domain": domain, "bucket": bucket})
 
 
 @admin_api_bp.route("/website-domains/<domain>", methods=["DELETE"])
@@ -765,6 +771,7 @@ def delete_website_domain(domain: str):
         return error
     if not current_app.config.get("WEBSITE_HOSTING_ENABLED", False):
         return _json_error("InvalidRequest", "Website hosting is not enabled", 400)
+    domain = normalize_domain(domain)
     if not _website_domains().delete_mapping(domain):
         return _json_error("NotFound", f"No mapping found for domain '{domain}'", 404)
     logger.info("Website domain mapping deleted: %s", domain)

app/ui.py

@@ -51,6 +51,7 @@ from .s3_client import (
 from .secret_store import EphemeralSecretStore
 from .site_registry import SiteRegistry, SiteInfo, PeerSite
 from .storage import ObjectStorage, StorageError
+from .website_domains import normalize_domain, is_valid_domain
 
 ui_bp = Blueprint("ui", __name__, template_folder="../templates", url_prefix="/ui")
 
@@ -2400,7 +2401,7 @@ def create_website_domain():
         flash("Website hosting is not enabled", "warning")
         return redirect(url_for("ui.buckets_overview"))
 
-    domain = (request.form.get("domain") or "").strip().lower()
+    domain = normalize_domain(request.form.get("domain") or "")
     bucket = (request.form.get("bucket") or "").strip()
 
     if not domain:
@@ -2409,6 +2410,12 @@ def create_website_domain():
         flash("Domain is required", "danger")
         return redirect(url_for("ui.website_domains_dashboard"))
 
+    if not is_valid_domain(domain):
+        if _wants_json():
+            return jsonify({"error": f"Invalid domain format: '{domain}'"}), 400
+        flash(f"Invalid domain format: '{domain}'. Use a hostname like www.example.com", "danger")
+        return redirect(url_for("ui.website_domains_dashboard"))
+
     if not bucket:
         if _wants_json():
             return jsonify({"error": "Bucket is required"}), 400
@@ -2447,6 +2454,7 @@ def update_website_domain(domain: str):
         flash("Access denied", "danger")
         return redirect(url_for("ui.website_domains_dashboard"))
 
+    domain = normalize_domain(domain)
     bucket = (request.form.get("bucket") or "").strip()
     if not bucket:
         if _wants_json():
@@ -2462,9 +2470,14 @@ def update_website_domain(domain: str):
         return redirect(url_for("ui.website_domains_dashboard"))
 
     store = current_app.extensions.get("website_domains")
+    if not store.get_bucket(domain):
+        if _wants_json():
+            return jsonify({"error": f"No mapping for domain '{domain}'"}), 404
+        flash(f"No mapping for domain '{domain}'", "danger")
+        return redirect(url_for("ui.website_domains_dashboard"))
     store.set_mapping(domain, bucket)
     if _wants_json():
-        return jsonify({"success": True, "domain": domain.lower(), "bucket": bucket})
+        return jsonify({"success": True, "domain": domain, "bucket": bucket})
     flash(f"Domain '{domain}' updated to bucket '{bucket}'", "success")
     return redirect(url_for("ui.website_domains_dashboard"))
 
@@ -2480,6 +2493,7 @@ def delete_website_domain(domain: str):
         flash("Access denied", "danger")
         return redirect(url_for("ui.website_domains_dashboard"))
 
+    domain = normalize_domain(domain)
     store = current_app.extensions.get("website_domains")
     if not store.delete_mapping(domain):
         if _wants_json():

app/website_domains.py

@@ -1,10 +1,34 @@
 from __future__ import annotations
 
 import json
+import re
 import threading
 from pathlib import Path
 from typing import Dict, List, Optional
 
+_DOMAIN_RE = re.compile(
+    r"^(?!-)[a-z0-9]([a-z0-9-]*[a-z0-9])?(\.[a-z0-9]([a-z0-9-]*[a-z0-9])?)*$"
+)
+
+
+def normalize_domain(raw: str) -> str:
+    raw = raw.strip().lower()
+    for prefix in ("https://", "http://"):
+        if raw.startswith(prefix):
+            raw = raw[len(prefix):]
+    raw = raw.split("/", 1)[0]
+    raw = raw.split("?", 1)[0]
+    raw = raw.split("#", 1)[0]
+    if ":" in raw:
+        raw = raw.rsplit(":", 1)[0]
+    return raw
+
+
+def is_valid_domain(domain: str) -> bool:
+    if not domain or len(domain) > 253:
+        return False
+    return bool(_DOMAIN_RE.match(domain))
+
 
 class WebsiteDomainStore:
     def __init__(self, config_path: Path) -> None: