203 lines
6.7 KiB
Python
203 lines
6.7 KiB
Python
"""Live DNS checks for company domain onboarding."""
|
|
|
|
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()
|
|
|
|
|
|
def _lookup_txt(name: str) -> list[str]:
|
|
try:
|
|
answers = dns.resolver.resolve(name, "TXT")
|
|
except (
|
|
dns.resolver.NXDOMAIN,
|
|
dns.resolver.NoAnswer,
|
|
dns.resolver.NoNameservers,
|
|
dns.exception.Timeout,
|
|
dns.resolver.LifetimeTimeout,
|
|
):
|
|
return []
|
|
records: list[str] = []
|
|
for answer in answers:
|
|
parts = []
|
|
for chunk in answer.strings:
|
|
if isinstance(chunk, bytes):
|
|
parts.append(chunk.decode("utf-8", errors="replace"))
|
|
else:
|
|
parts.append(str(chunk))
|
|
records.append("".join(parts))
|
|
return records
|
|
|
|
|
|
def _lookup_mx(domain_name: str) -> list[tuple[int, str]]:
|
|
try:
|
|
answers = dns.resolver.resolve(domain_name, "MX")
|
|
except (
|
|
dns.resolver.NXDOMAIN,
|
|
dns.resolver.NoAnswer,
|
|
dns.resolver.NoNameservers,
|
|
dns.exception.Timeout,
|
|
dns.resolver.LifetimeTimeout,
|
|
):
|
|
return []
|
|
hosts: list[tuple[int, str]] = []
|
|
for answer in answers:
|
|
host = str(answer.exchange).rstrip(".").lower()
|
|
hosts.append((int(answer.preference), host))
|
|
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)
|
|
if not hosts:
|
|
return False, f"No MX records found for {domain_name}."
|
|
hostnames = [host for _, host in hosts]
|
|
if expected in hostnames:
|
|
return True, f"MX points to {expected_host}."
|
|
return False, f"Expected MX host {expected_host}; found {', '.join(hostnames)}."
|
|
|
|
|
|
def verify_spf(domain_name: str, expected_include: str) -> tuple[bool, str]:
|
|
include_token = expected_include.rstrip(".").lower()
|
|
records = [_normalize_txt(record) for record in _lookup_txt(domain_name)]
|
|
spf_records = [record for record in records if record.startswith("v=spf1")]
|
|
if not spf_records:
|
|
return False, f"No SPF TXT record found at {domain_name}."
|
|
needle = f"include:{include_token}"
|
|
for record in spf_records:
|
|
if needle in record:
|
|
return True, f"SPF includes {expected_include}."
|
|
return False, f"SPF record must include {expected_include}."
|
|
|
|
|
|
def _lookup_a(name: str) -> list[str]:
|
|
try:
|
|
answers = dns.resolver.resolve(name, "A")
|
|
except (
|
|
dns.resolver.NXDOMAIN,
|
|
dns.resolver.NoAnswer,
|
|
dns.resolver.NoNameservers,
|
|
dns.exception.Timeout,
|
|
dns.resolver.LifetimeTimeout,
|
|
):
|
|
return []
|
|
return [str(answer).strip() for answer in answers]
|
|
|
|
|
|
def verify_platform_spf(platform_spf_host: str, mail_host: str) -> tuple[bool, str]:
|
|
"""Ensure the shared SPF include target exists and authorizes the mail host."""
|
|
records = [_normalize_txt(record) for record in _lookup_txt(platform_spf_host)]
|
|
spf_records = [record for record in records if record.startswith("v=spf1")]
|
|
if not spf_records:
|
|
return (
|
|
False,
|
|
f"No SPF TXT at {platform_spf_host}. On the platform domain add Host _spf with "
|
|
f"v=spf1 ip4:<mail-server-ip> ~all.",
|
|
)
|
|
|
|
spf_value = spf_records[0]
|
|
mail_ips = _lookup_a(mail_host.rstrip("."))
|
|
for ip in mail_ips:
|
|
if f"ip4:{ip}" in spf_value or f"ip6:{ip}" in spf_value:
|
|
return True, f"{platform_spf_host} authorizes {mail_host} ({ip})."
|
|
|
|
if "ip4:" in spf_value or "ip6:" in spf_value or " a " in f" {spf_value} ":
|
|
return True, f"{platform_spf_host} SPF record is present."
|
|
|
|
return (
|
|
False,
|
|
f"{platform_spf_host} exists but does not appear to authorize the mail server.",
|
|
)
|
|
|
|
|
|
def verify_dkim(
|
|
domain_name: str,
|
|
selector: str,
|
|
expected_public_key: str,
|
|
) -> tuple[bool, str]:
|
|
name = f"{selector}._domainkey.{domain_name}"
|
|
records = [_normalize_txt(record) for record in _lookup_txt(name)]
|
|
if not records:
|
|
return False, f"No DKIM TXT record found at {name}."
|
|
key_token = expected_public_key.strip().lower()
|
|
for record in records:
|
|
if key_token in record.replace(" ", ""):
|
|
return True, "DKIM public key matches."
|
|
return False, f"DKIM TXT at {selector}._domainkey is missing the expected public key."
|
|
|
|
|
|
def verify_dmarc(domain_name: str, expected_value: str) -> tuple[bool, str]:
|
|
name = f"_dmarc.{domain_name}"
|
|
records = [_normalize_txt(record) for record in _lookup_txt(name)]
|
|
if not records:
|
|
return False, f"No DMARC TXT record found at {name}."
|
|
expected = _normalize_txt(expected_value)
|
|
for record in records:
|
|
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(
|
|
*,
|
|
domain_name: str,
|
|
shared_mail_host: str,
|
|
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)),
|
|
(
|
|
"dkim",
|
|
*verify_dkim(domain_name, dkim_selector, dkim_public_key),
|
|
),
|
|
("dmarc", *verify_dmarc(domain_name, dmarc_expected)),
|
|
]
|
|
return [
|
|
{
|
|
"purpose": purpose,
|
|
"verified": verified,
|
|
"message": message,
|
|
}
|
|
for purpose, verified, message in checks
|
|
]
|