feat: Update CustomDomainSettings to require CUSTOM_DOMAINS_ prefix for environment variables and add unit tests for validation

Author: Zingzy

config.py

@@ -125,9 +125,17 @@ class CustomDomainSettings(BaseSettings):
     rollout flag flips. ``enabled`` is the master switch consulted by the
     service layer (PR5+); the data plumbing (schema, repo, wiring) lands
     even when False so the rollout has a clean code path to flip.
+
+    All env vars must be prefixed ``CUSTOM_DOMAINS_`` so generic names
+    like ``ENABLED`` or ``MAX_PER_USER`` set elsewhere in the deploy
+    environment don't accidentally configure this feature.
     """
 
-    model_config = SettingsConfigDict(env_file=".env", extra="ignore")
+    model_config = SettingsConfigDict(
+        env_file=".env",
+        extra="ignore",
+        env_prefix="CUSTOM_DOMAINS_",
+    )
 
     # Master switch consulted by CustomDomainService. Until True, every
     # public method short-circuits with DomainQuotaExceededError or similar.

repositories/custom_domain_repository.py

@@ -21,22 +21,36 @@
 log = get_logger(__name__)
 
 
+def _canonical(fqdn: str) -> str:
+    """Cheap normalisation for lookup parameters.
+
+    Persisted docs are validated through ``normalise_fqdn`` (strict) before
+    insert, so they're already canonical. Lookups normalise too — different
+    callers (DTO-validated input, raw middleware string, ops mongosh
+    input) reach the same row regardless of case or trailing dots. Kept
+    cheap (no regex) because lookup syntax is the caller's job, not ours.
+    """
+    return str(fqdn).strip().lower().rstrip(".")
+
+
 class CustomDomainRepository(BaseRepository[CustomDomainDoc]):
     async def find_by_id(self, domain_id: ObjectId) -> CustomDomainDoc | None:
         """Find a domain by its ObjectId."""
         return await self._find_one({"_id": domain_id})
 
     async def find_by_fqdn(self, fqdn: str) -> CustomDomainDoc | None:
         """Find a domain by fqdn (any status). Used by uniqueness checks."""
-        return await self._find_one({"fqdn": fqdn})
+        return await self._find_one({"fqdn": _canonical(fqdn)})
 
     async def find_active_by_fqdn(self, fqdn: str) -> CustomDomainDoc | None:
         """Find a domain by fqdn, scoped to ACTIVE only.
 
         Used by the Caddy ask endpoint — we only mint certs for verified,
         currently-active domains.
         """
-        return await self._find_one({"fqdn": fqdn, "status": DomainStatus.ACTIVE})
+        return await self._find_one(
+            {"fqdn": _canonical(fqdn), "status": DomainStatus.ACTIVE}
+        )
 
     async def list_by_owner(
         self,

schemas/dto/requests/custom_domain.py

@@ -2,35 +2,13 @@
 
 from __future__ import annotations
 
-import re
 from typing import Any
 
 from pydantic import Field, field_validator
 
 from schemas.dto.base import RequestBase
 from schemas.enums.domain_status import VerificationMethod
-
-# Same regex as schemas.models.custom_domain — duplicated here so DTO
-# validation rejects bad input at the API boundary without depending on
-# the document model.
-_HOSTNAME_RE = re.compile(
-    r"^(?=.{1,253}$)"
-    r"(?:[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?\.)+"
-    r"[a-z]{2,63}$"
-)
-
-
-def _normalise_fqdn(v: Any) -> str:
-    if v is None:
-        raise ValueError("fqdn is required")
-    normalised = str(v).strip().lower().rstrip(".")
-    if not normalised:
-        raise ValueError("fqdn is required")
-    if re.search(r"[\x00-\x1F\x7F-\x9F<>\"'`\\]", normalised):
-        raise ValueError("fqdn contains forbidden characters")
-    if not _HOSTNAME_RE.match(normalised):
-        raise ValueError("fqdn does not look like a valid hostname")
-    return normalised
+from shared.url_utils import normalise_fqdn
 
 
 class CreateCustomDomainRequest(RequestBase):
@@ -55,7 +33,7 @@ class CreateCustomDomainRequest(RequestBase):
     @field_validator("fqdn", mode="before")
     @classmethod
     def _validate_fqdn(cls, v: Any) -> str:
-        return _normalise_fqdn(v)
+        return normalise_fqdn(v)
 
     @field_validator("verification_method", mode="before")
     @classmethod

schemas/models/custom_domain.py

@@ -13,24 +13,14 @@
 
 from __future__ import annotations
 
-import re
 from datetime import datetime
 from typing import Any
 
 from pydantic import field_validator
 
 from schemas.enums.domain_status import DomainStatus, VerificationMethod
 from schemas.models.base import MongoBaseModel, PyObjectId
-
-# RFC 1035 hostname: labels of [a-z0-9-], 1-63 chars each, separated by dots,
-# total length ≤ 253. Trailing dot stripped before validation. Allows internal
-# uppercase (we lowercase ourselves) and rejects leading/trailing hyphens per
-# label.
-_HOSTNAME_RE = re.compile(
-    r"^(?=.{1,253}$)"
-    r"(?:[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?\.)+"
-    r"[a-z]{2,63}$"
-)
+from shared.url_utils import normalise_fqdn
 
 
 class CustomDomainDoc(MongoBaseModel):
@@ -61,19 +51,8 @@ class CustomDomainDoc(MongoBaseModel):
 
     @field_validator("fqdn", mode="before")
     @classmethod
-    def _normalise_fqdn(cls, v: Any) -> str:
-        if v is None:
-            raise ValueError("fqdn is required")
-        normalised = str(v).strip().lower().rstrip(".")
-        if not normalised:
-            raise ValueError("fqdn is required")
-        # Reject control + HTML metacharacters explicitly so a malformed input
-        # can't sneak past the regex via Unicode tricks.
-        if re.search(r"[\x00-\x1F\x7F-\x9F<>\"'`\\]", normalised):
-            raise ValueError(f"fqdn contains forbidden characters: {v!r}")
-        if not _HOSTNAME_RE.match(normalised):
-            raise ValueError(f"fqdn does not look like a valid hostname: {v!r}")
-        return normalised
+    def _normalise(cls, v: Any) -> str:
+        return normalise_fqdn(v)
 
 
 # Convenience: the set of legal state transitions, used by the service to

services/verifiers/a_record_verifier.py

@@ -15,11 +15,8 @@
 import dns.exception
 import dns.resolver
 
-from infrastructure.logging import get_logger
 from services.verifiers.protocol import DomainVerifier, VerificationResult
 
-log = get_logger(__name__)
-
 _DNS_TIMEOUT_SECS = 3.0
 
 

services/verifiers/cname_verifier.py

@@ -11,14 +11,10 @@
 
 import dns.asyncresolver
 import dns.exception
-import dns.rdatatype
 import dns.resolver
 
-from infrastructure.logging import get_logger
 from services.verifiers.protocol import DomainVerifier, VerificationResult
 
-log = get_logger(__name__)
-
 # Per-query DNS timeout. Kept short so a single slow nameserver can't stall
 # the worker; total budget = timeout * dnspython retries (default 2).
 _DNS_TIMEOUT_SECS = 3.0

services/verifiers/txt_challenge_verifier.py

@@ -13,11 +13,8 @@
 import dns.exception
 import dns.resolver
 
-from infrastructure.logging import get_logger
 from services.verifiers.protocol import DomainVerifier, VerificationResult
 
-log = get_logger(__name__)
-
 _DNS_TIMEOUT_SECS = 3.0
 _CHALLENGE_PREFIX = "_spoo-challenge"
 

shared/url_utils.py

@@ -2,8 +2,21 @@
 
 from __future__ import annotations
 
+import re
 from urllib.parse import urlparse
 
+# RFC 1035 hostname matcher used by the custom-domains code path.
+# Labels: 1-63 chars, [a-z0-9-], no leading/trailing hyphen.
+# Total length: ≤ 253.
+# TLD: either ≥2 alpha chars OR an ASCII-encoded punycode label (``xn--…``)
+# so internationalised TLDs (.中国 → ``xn--fiqs8s``) are accepted.
+_HOSTNAME_RE = re.compile(
+    r"^(?=.{1,253}$)"
+    r"(?:[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?\.)+"
+    r"(?:[a-z]{2,63}|xn--[a-z0-9-]{1,59})$"
+)
+_FORBIDDEN_CHARS = re.compile(r"[\x00-\x1F\x7F-\x9F<>\"'`\\]")
+
 
 def extract_hostname(url: str | None) -> str | None:
     """Return hostname from URL, or None if unparseable."""
@@ -30,3 +43,26 @@ def extract_fqdn(url: str) -> str:
     if not host:
         return "localhost"
     return host.lower().rstrip(".")
+
+
+def normalise_fqdn(value: object) -> str:
+    """Strict canonical form for custom-domain fqdns.
+
+    Strips whitespace, lowercases, drops a trailing dot, and validates
+    against RFC 1035 hostname syntax (with punycode TLD support). Raises
+    ``ValueError`` for empty / bad-character / malformed input.
+
+    Single source of truth — used by the document model, request DTO,
+    AND the repository so a normalised lookup never misses because the
+    persisted form drifted from the input form.
+    """
+    if value is None:
+        raise ValueError("fqdn is required")
+    normalised = str(value).strip().lower().rstrip(".")
+    if not normalised:
+        raise ValueError("fqdn is required")
+    if _FORBIDDEN_CHARS.search(normalised):
+        raise ValueError(f"fqdn contains forbidden characters: {value!r}")
+    if not _HOSTNAME_RE.match(normalised):
+        raise ValueError(f"fqdn does not look like a valid hostname: {value!r}")
+    return normalised

tests/unit/repositories/test_custom_domain_repository.py

@@ -40,6 +40,30 @@ async def test_find_by_fqdn_returns_doc(self):
         assert doc.fqdn == "links.example.com"
         col.find_one.assert_awaited_once_with({"fqdn": "links.example.com"})
 
+    @pytest.mark.asyncio
+    async def test_find_by_fqdn_normalises_lookup_input(self):
+        # A caller passing an uppercased, trailing-dot, padded variant must
+        # still find the canonical row — defends against input drift between
+        # the writer (DTO-validated) and ad-hoc lookup paths.
+        col = AsyncMock()
+        col.find_one = AsyncMock(return_value=None)
+        col.name = "custom_domains"
+        repo = CustomDomainRepository(col)
+
+        await repo.find_by_fqdn("  LINKS.Example.COM.  ")
+        col.find_one.assert_awaited_once_with({"fqdn": "links.example.com"})
+
+    @pytest.mark.asyncio
+    async def test_find_active_by_fqdn_normalises_lookup_input(self):
+        col = AsyncMock()
+        col.find_one = AsyncMock(return_value=None)
+        col.name = "custom_domains"
+        repo = CustomDomainRepository(col)
+
+        await repo.find_active_by_fqdn("LINKS.Example.COM.")
+        args = col.find_one.call_args.args[0]
+        assert args["fqdn"] == "links.example.com"
+
     @pytest.mark.asyncio
     async def test_find_active_by_fqdn_scopes_to_active(self):
         col = AsyncMock()

tests/unit/shared/test_url_utils.py

@@ -1,6 +1,8 @@
-"""Unit tests for shared/url_utils.py — extract_hostname + extract_fqdn."""
+"""Unit tests for shared/url_utils.py — extract_hostname + extract_fqdn + normalise_fqdn."""
 
-from shared.url_utils import extract_fqdn, extract_hostname
+import pytest
+
+from shared.url_utils import extract_fqdn, extract_hostname, normalise_fqdn
 
 
 class TestExtractHostname:
@@ -49,3 +51,45 @@ def test_idempotent(self):
         # so the cache key, the seeded custom_domains row, and the request
         # middleware all agree on the canonical form.
         assert extract_fqdn("HTTPS://Spoo.Me./") == extract_fqdn("https://spoo.me")
+
+
+class TestNormaliseFqdn:
+    @pytest.mark.parametrize(
+        "value, expected",
+        [
+            ("links.acme.com", "links.acme.com"),
+            ("LINKS.ACME.COM", "links.acme.com"),
+            ("  links.acme.com  ", "links.acme.com"),
+            ("links.acme.com.", "links.acme.com"),
+            ("acme.co", "acme.co"),
+            # Punycode TLD (encoded `.中国`) — required for IDN custom domains.
+            ("links.xn--fiqs8s", "links.xn--fiqs8s"),
+            # Multi-level subdomain
+            ("a.b.c.example.com", "a.b.c.example.com"),
+        ],
+    )
+    def test_accepts_valid_inputs(self, value, expected):
+        assert normalise_fqdn(value) == expected
+
+    @pytest.mark.parametrize(
+        "value",
+        [
+            None,
+            "",
+            "   ",
+            "no_underscores_allowed.com",
+            "-leading-hyphen.com",
+            "trailing-hyphen-.com",
+            "single-label",
+            "two..consecutive.dots.com",
+            "evil<script>.com",
+            "evil`backtick.com",
+            "evil\\backslash.com",
+            "evil\x00null.com",
+            "a" * 64 + ".com",  # label > 63 chars
+            "a" * 254 + ".com",  # total > 253 chars
+        ],
+    )
+    def test_rejects_invalid_inputs(self, value):
+        with pytest.raises(ValueError):
+            normalise_fqdn(value)