domain migration

backend/app/services/dns_verification.py

@@ -5,6 +5,8 @@ from __future__ import annotations
 import dns.exception
 import dns.resolver
 
+OWNERSHIP_PREFIX = "agmail-verify="
+
 
 def _normalize_txt(value: str) -> str:
     return " ".join(value.split()).strip().lower()
@@ -51,6 +53,29 @@ def _lookup_mx(domain_name: str) -> list[tuple[int, str]]:
     return hosts
 
 
+def ownership_record_value(token: str) -> str:
+    return f"{OWNERSHIP_PREFIX}{token}"
+
+
+def ownership_token_from_value(value: str) -> str:
+    normalized = value.strip()
+    if normalized.lower().startswith(OWNERSHIP_PREFIX):
+        return normalized[len(OWNERSHIP_PREFIX) :]
+    return normalized
+
+
+def verify_ownership(domain_name: str, verification_token: str) -> tuple[bool, str]:
+    name = f"_agmail-verify.{domain_name}"
+    expected = _normalize_txt(ownership_record_value(verification_token))
+    records = [_normalize_txt(record) for record in _lookup_txt(name)]
+    if expected in records:
+        return True, "Domain ownership verified via TXT record."
+    return (
+        False,
+        f"Add TXT at _agmail-verify.{domain_name} with value {ownership_record_value(verification_token)}",
+    )
+
+
 def verify_mx(domain_name: str, expected_host: str) -> tuple[bool, str]:
     expected = expected_host.rstrip(".").lower()
     hosts = _lookup_mx(domain_name)
@@ -96,7 +121,7 @@ def verify_platform_spf(platform_spf_host: str, mail_host: str) -> tuple[bool, s
     if not spf_records:
         return (
             False,
-            f"No SPF TXT at {platform_spf_host}. On getall.help add Host _spf with "
+            f"No SPF TXT at {platform_spf_host}. On the platform domain add Host _spf with "
             f"v=spf1 ip4:<mail-server-ip> ~all.",
         )
 
@@ -138,10 +163,13 @@ def verify_dmarc(domain_name: str, expected_value: str) -> tuple[bool, str]:
         return False, f"No DMARC TXT record found at {name}."
     expected = _normalize_txt(expected_value)
     for record in records:
-        if record == expected or record.startswith("v=dmarc1"):
-            if "v=dmarc1" in record:
-                return True, "DMARC record is present."
-    return False, "DMARC TXT record is missing or incomplete."
+        if record == expected:
+            return True, "DMARC record matches the expected value."
+    return (
+        False,
+        "DMARC TXT must match the value shown in setup exactly "
+        "(do not rely on an existing third-party DMARC record).",
+    )
 
 
 def verify_company_dns(
@@ -151,15 +179,14 @@ def verify_company_dns(
     dkim_selector: str,
     dkim_public_key: str,
     platform_spf_include: str,
+    verification_token: str,
 ) -> list[dict]:
+    """Customer-domain checks only (ownership + mail authentication)."""
     dmarc_expected = f"v=DMARC1; p=quarantine; rua=mailto:dmarc@{domain_name}"
     checks = [
+        ("ownership", *verify_ownership(domain_name, verification_token)),
         ("mx", *verify_mx(domain_name, shared_mail_host)),
         ("spf", *verify_spf(domain_name, platform_spf_include)),
-        (
-            "platform_spf",
-            *verify_platform_spf(platform_spf_include, shared_mail_host),
-        ),
         (
             "dkim",
             *verify_dkim(domain_name, dkim_selector, dkim_public_key),

backend/app/services/domain_policy.py

@@ -0,0 +1,49 @@
+"""Company domain registration policy."""
+
+from __future__ import annotations
+
+from app.services.domain_provisioning import normalize_domain, validate_domain
+
+# Well-known domains that must never be used for company onboarding.
+BLOCKED_COMPANY_DOMAINS = frozenset(
+    {
+        "google.com",
+        "gmail.com",
+        "yahoo.com",
+        "outlook.com",
+        "hotmail.com",
+        "live.com",
+        "icloud.com",
+        "apple.com",
+        "microsoft.com",
+        "amazon.com",
+        "facebook.com",
+        "meta.com",
+        "linkedin.com",
+        "proton.me",
+        "protonmail.com",
+        "aol.com",
+        "zoho.com",
+        "fastmail.com",
+        "yandex.com",
+        "mail.ru",
+        "cloudflare.com",
+        "github.com",
+    }
+)
+
+
+def assert_company_domain_allowed(domain: str, platform_domain: str) -> str:
+    """Validate format and reject reserved or platform domains."""
+    normalized = validate_domain(domain)
+    if normalized == normalize_domain(platform_domain):
+        raise ValueError(
+            f"You cannot register {platform_domain} as a company domain. "
+            "Use a domain your organization controls."
+        )
+    if normalized in BLOCKED_COMPANY_DOMAINS:
+        raise ValueError(
+            "That domain cannot be used for company signup. "
+            "Enter a domain you control and can edit in DNS."
+        )
+    return normalized

backend/app/services/domain_provisioning.py

@@ -94,11 +94,21 @@ def build_company_dns_records(
     dkim_selector: str,
     dkim_public_key: str,
     platform_spf_include: str,
+    verification_token: str,
 ) -> list[dict]:
-    """Zoho-style records: shared MX host, SPF include, per-domain DKIM/DMARC."""
+    """Zoho-style records: ownership proof, shared MX host, SPF include, DKIM/DMARC."""
+    from app.services.dns_verification import ownership_record_value
+
     dkim_value = f"v=DKIM1; k=rsa; p={dkim_public_key}"
     dmarc_value = f"v=DMARC1; p=quarantine; rua=mailto:dmarc@{domain_name}"
     return [
+        {
+            "record_type": "TXT",
+            "host": namecheap_host("_agmail-verify", domain_name),
+            "value": ownership_record_value(verification_token),
+            "purpose": "ownership",
+            "ttl": 3600,
+        },
         {
             "record_type": "MX",
             "host": namecheap_host("@", domain_name),
@@ -273,12 +283,14 @@ async def provision_mail_domain(
     private_key, public_key = generate_dkim_keys()
     if domain_type == "company":
         mail_host = settings.smtp_public_host
+        verification_token = secrets.token_urlsafe(24)
         dns_records = build_company_dns_records(
             normalized,
             settings.smtp_public_host,
             dkim_selector,
             public_key,
             settings.platform_spf_include,
+            verification_token,
         )
     else:
         dns_records = build_dns_records(
@@ -331,6 +343,48 @@ async def activate_company_domain(session: AsyncSession, domain: MailDomain) ->
     return domain
 
 
+async def ensure_ownership_dns_record(session: AsyncSession, domain: MailDomain) -> str:
+    """Return the ownership verification token, creating a DNS record if missing."""
+    from app.services.dns_verification import ownership_record_value, ownership_token_from_value
+
+    existing = next(
+        (record for record in domain.dns_records if record.purpose == "ownership"),
+        None,
+    )
+    if existing:
+        return ownership_token_from_value(existing.value)
+
+    token = secrets.token_urlsafe(24)
+    session.add(
+        DomainDnsRecord(
+            domain_id=domain.id,
+            record_type="TXT",
+            host=namecheap_host("_agmail-verify", domain.domain_name),
+            value=ownership_record_value(token),
+            purpose="ownership",
+            ttl=3600,
+        )
+    )
+    await session.flush()
+    return token
+
+
+async def account_is_domain_owner(
+    session: AsyncSession,
+    account_id,
+    company_domain_name: str | None,
+) -> bool:
+    if not company_domain_name:
+        return False
+    domain_id = await session.scalar(
+        select(MailDomain.id).where(
+            MailDomain.domain_name == company_domain_name,
+            MailDomain.account_id == account_id,
+        )
+    )
+    return domain_id is not None
+
+
 async def get_account_company_domain(
     session: AsyncSession,
     account_id,