465 lines
14 KiB
Python
465 lines
14 KiB
Python
import ipaddress
|
|
import re
|
|
import secrets
|
|
import string
|
|
from pathlib import Path
|
|
|
|
from cryptography.hazmat.primitives import serialization
|
|
from cryptography.hazmat.primitives.asymmetric import rsa
|
|
from pwdlib import PasswordHash
|
|
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
from sqlalchemy.orm import selectinload
|
|
|
|
from app.config import get_settings
|
|
from app.models import DomainDnsRecord, MailDomain, SmtpCredential
|
|
|
|
password_hasher = PasswordHash.recommended()
|
|
domain_pattern = re.compile(
|
|
r"^(?=.{1,253}$)(?:[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?\.)+"
|
|
r"[a-z]{2,63}$"
|
|
)
|
|
|
|
|
|
def normalize_domain(value: str) -> str:
|
|
return value.strip().lower().removeprefix("@")
|
|
|
|
|
|
def validate_domain(value: str) -> str:
|
|
domain = normalize_domain(value)
|
|
if not domain_pattern.fullmatch(domain):
|
|
raise ValueError("Enter a valid domain name, such as example.com.")
|
|
return domain
|
|
|
|
|
|
def generate_dkim_keys() -> tuple[str, str]:
|
|
private_key = rsa.generate_private_key(public_exponent=65537, key_size=2048)
|
|
private_pem = private_key.private_bytes(
|
|
encoding=serialization.Encoding.PEM,
|
|
format=serialization.PrivateFormat.PKCS8,
|
|
encryption_algorithm=serialization.NoEncryption(),
|
|
).decode()
|
|
public_pem = (
|
|
private_key.public_key()
|
|
.public_bytes(
|
|
encoding=serialization.Encoding.PEM,
|
|
format=serialization.PublicFormat.SubjectPublicKeyInfo,
|
|
)
|
|
.decode()
|
|
)
|
|
public_key = "".join(
|
|
line
|
|
for line in public_pem.splitlines()
|
|
if not line.startswith("-----")
|
|
)
|
|
return private_pem, public_key
|
|
|
|
|
|
def namecheap_host(host: str, domain_name: str) -> str:
|
|
"""Strip the zone suffix so records match Namecheap's Host field."""
|
|
if host in {"@", "*"}:
|
|
return host
|
|
suffix = f".{domain_name}"
|
|
if host.endswith(suffix):
|
|
return host[: -len(suffix)]
|
|
return host
|
|
|
|
|
|
def spf_ip_mechanism(mail_server_ip: str | None) -> str | None:
|
|
if not mail_server_ip or mail_server_ip == "YOUR_MAIL_SERVER_IP":
|
|
return None
|
|
try:
|
|
address = ipaddress.ip_address(mail_server_ip.strip())
|
|
except ValueError:
|
|
return None
|
|
return f"ip6:{address}" if address.version == 6 else f"ip4:{address}"
|
|
|
|
|
|
def build_spf_value(mail_server_ip: str | None = None) -> str:
|
|
parts = ["v=spf1", "mx", "a"]
|
|
ip_mechanism = spf_ip_mechanism(mail_server_ip)
|
|
if ip_mechanism:
|
|
parts.append(ip_mechanism)
|
|
parts.append("~all")
|
|
return " ".join(parts)
|
|
|
|
|
|
def build_shared_mx_spf_value(platform_spf_include: str) -> str:
|
|
return f"v=spf1 include:{platform_spf_include} ~all"
|
|
|
|
|
|
def build_company_dns_records(
|
|
domain_name: str,
|
|
shared_mail_host: str,
|
|
dkim_selector: str,
|
|
dkim_public_key: str,
|
|
platform_spf_include: str,
|
|
verification_token: str,
|
|
) -> list[dict]:
|
|
"""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),
|
|
"value": shared_mail_host,
|
|
"priority": 10,
|
|
"purpose": "mx",
|
|
"ttl": 3600,
|
|
},
|
|
{
|
|
"record_type": "TXT",
|
|
"host": namecheap_host("@", domain_name),
|
|
"value": build_shared_mx_spf_value(platform_spf_include),
|
|
"purpose": "spf",
|
|
"ttl": 3600,
|
|
},
|
|
{
|
|
"record_type": "TXT",
|
|
"host": namecheap_host(f"{dkim_selector}._domainkey", domain_name),
|
|
"value": dkim_value,
|
|
"purpose": "dkim",
|
|
"ttl": 3600,
|
|
},
|
|
{
|
|
"record_type": "TXT",
|
|
"host": namecheap_host("_dmarc", domain_name),
|
|
"value": dmarc_value,
|
|
"purpose": "dmarc",
|
|
"ttl": 3600,
|
|
},
|
|
]
|
|
|
|
|
|
def build_dns_records(
|
|
domain_name: str,
|
|
mail_host: str,
|
|
dkim_selector: str,
|
|
dkim_public_key: str,
|
|
mail_server_ip: str | None = None,
|
|
) -> list[dict]:
|
|
dkim_value = f"v=DKIM1; k=rsa; p={dkim_public_key}"
|
|
dmarc_value = f"v=DMARC1; p=quarantine; rua=mailto:dmarc@{domain_name}"
|
|
mail_subdomain = mail_host.removesuffix(f".{domain_name}")
|
|
|
|
return [
|
|
{
|
|
"record_type": "MX",
|
|
"host": namecheap_host("@", domain_name),
|
|
"value": mail_host,
|
|
"priority": 10,
|
|
"purpose": "mx",
|
|
"ttl": 3600,
|
|
},
|
|
{
|
|
"record_type": "A",
|
|
"host": namecheap_host(mail_subdomain, domain_name),
|
|
"value": mail_server_ip or "YOUR_MAIL_SERVER_IP",
|
|
"purpose": "mail_host",
|
|
"ttl": 3600,
|
|
},
|
|
{
|
|
"record_type": "TXT",
|
|
"host": namecheap_host("@", domain_name),
|
|
"value": build_spf_value(mail_server_ip),
|
|
"purpose": "spf",
|
|
"ttl": 3600,
|
|
},
|
|
{
|
|
"record_type": "TXT",
|
|
"host": namecheap_host(f"{dkim_selector}._domainkey", domain_name),
|
|
"value": dkim_value,
|
|
"purpose": "dkim",
|
|
"ttl": 3600,
|
|
},
|
|
{
|
|
"record_type": "TXT",
|
|
"host": namecheap_host("_dmarc", domain_name),
|
|
"value": dmarc_value,
|
|
"purpose": "dmarc",
|
|
"ttl": 3600,
|
|
},
|
|
]
|
|
|
|
|
|
def build_deployment_dns_records(
|
|
*,
|
|
domain_name: str,
|
|
mail_server_ip: str,
|
|
dkim_selector: str,
|
|
dkim_public_key: str | None,
|
|
app_public_host: str | None = None,
|
|
) -> dict[str, list[dict]]:
|
|
"""Split records into inbound delivery vs outbound authentication."""
|
|
mail_host = f"mail.{domain_name}"
|
|
dkim_key = dkim_public_key or (
|
|
"REGISTER_DOMAIN_TO_GENERATE_DKIM_PUBLIC_KEY"
|
|
)
|
|
records = build_dns_records(
|
|
domain_name,
|
|
mail_host,
|
|
dkim_selector,
|
|
dkim_key,
|
|
mail_server_ip=mail_server_ip,
|
|
)
|
|
by_purpose = {record["purpose"]: record for record in records}
|
|
inbound = [by_purpose["mx"], by_purpose["mail_host"]]
|
|
if app_public_host:
|
|
inbound.append(
|
|
{
|
|
"record_type": "A",
|
|
"host": namecheap_host(app_public_host, domain_name),
|
|
"value": mail_server_ip,
|
|
"purpose": "app_host",
|
|
"ttl": 3600,
|
|
}
|
|
)
|
|
outbound = [by_purpose["spf"], by_purpose["dkim"], by_purpose["dmarc"]]
|
|
ip_mechanism = spf_ip_mechanism(mail_server_ip)
|
|
if ip_mechanism:
|
|
outbound.insert(
|
|
0,
|
|
{
|
|
"record_type": "TXT",
|
|
"host": namecheap_host("_spf", domain_name),
|
|
"value": f"v=spf1 {ip_mechanism} ~all",
|
|
"purpose": "platform_spf",
|
|
"ttl": 3600,
|
|
},
|
|
)
|
|
return {"inbound": inbound, "outbound": outbound}
|
|
|
|
|
|
def generate_smtp_username(domain_name: str) -> str:
|
|
slug = domain_name.replace(".", "-")
|
|
token = secrets.token_hex(4)
|
|
return f"smtp-{slug}-{token}"
|
|
|
|
|
|
def generate_smtp_password(length: int = 24) -> str:
|
|
alphabet = string.ascii_letters + string.digits
|
|
return "".join(secrets.choice(alphabet) for _ in range(length))
|
|
|
|
|
|
def record_for_db(record: dict) -> dict:
|
|
allowed = {"record_type", "host", "value", "purpose", "ttl"}
|
|
return {key: record[key] for key in allowed if key in record}
|
|
|
|
|
|
async def get_domain_by_name(
|
|
session: AsyncSession,
|
|
domain_name: str,
|
|
) -> MailDomain | None:
|
|
return await session.scalar(
|
|
select(MailDomain).where(MailDomain.domain_name == domain_name)
|
|
)
|
|
|
|
|
|
async def provision_mail_domain(
|
|
session: AsyncSession,
|
|
*,
|
|
domain_name: str,
|
|
domain_type: str,
|
|
account_id=None,
|
|
) -> MailDomain:
|
|
settings = get_settings()
|
|
normalized = validate_domain(domain_name)
|
|
existing = await get_domain_by_name(session, normalized)
|
|
if existing:
|
|
return existing
|
|
|
|
mail_host = f"mail.{normalized}"
|
|
dkim_selector = settings.dkim_selector
|
|
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(
|
|
normalized,
|
|
mail_host,
|
|
dkim_selector,
|
|
public_key,
|
|
)
|
|
domain = MailDomain(
|
|
domain_name=normalized,
|
|
domain_type=domain_type,
|
|
account_id=account_id,
|
|
mail_host=mail_host,
|
|
dkim_selector=dkim_selector,
|
|
dkim_private_key=private_key,
|
|
dkim_public_key=public_key,
|
|
status="pending_dns",
|
|
)
|
|
session.add(domain)
|
|
await session.flush()
|
|
|
|
for record in dns_records:
|
|
session.add(DomainDnsRecord(domain_id=domain.id, **record_for_db(record)))
|
|
|
|
await session.flush()
|
|
return domain
|
|
|
|
|
|
def register_relay_domain(domain_name: str) -> None:
|
|
"""Append a verified company domain for Postfix inbound relay."""
|
|
settings = get_settings()
|
|
normalized = validate_domain(domain_name)
|
|
path = Path(settings.relay_domains_file)
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
existing = {
|
|
line.strip().lower()
|
|
for line in path.read_text(encoding="utf-8").splitlines()
|
|
if line.strip()
|
|
} if path.exists() else set()
|
|
if normalized in existing:
|
|
return
|
|
with path.open("a", encoding="utf-8") as handle:
|
|
handle.write(f"{normalized}\n")
|
|
|
|
|
|
async def activate_company_domain(session: AsyncSession, domain: MailDomain) -> MailDomain:
|
|
domain.status = "active"
|
|
register_relay_domain(domain.domain_name)
|
|
await session.flush()
|
|
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_owned_domain(
|
|
session: AsyncSession,
|
|
domain_id,
|
|
account_id,
|
|
) -> MailDomain | None:
|
|
return await session.scalar(
|
|
select(MailDomain)
|
|
.options(
|
|
selectinload(MailDomain.dns_records),
|
|
selectinload(MailDomain.smtp_credentials),
|
|
)
|
|
.where(
|
|
MailDomain.id == domain_id,
|
|
MailDomain.account_id == account_id,
|
|
)
|
|
)
|
|
|
|
|
|
async def list_owned_domains(
|
|
session: AsyncSession,
|
|
account_id,
|
|
) -> list[MailDomain]:
|
|
result = await session.scalars(
|
|
select(MailDomain)
|
|
.options(selectinload(MailDomain.dns_records))
|
|
.where(MailDomain.account_id == account_id)
|
|
.order_by(MailDomain.created_at.asc())
|
|
)
|
|
return list(result.all())
|
|
|
|
|
|
async def get_account_company_domain(
|
|
session: AsyncSession,
|
|
account_id,
|
|
company_domain_name: str | None,
|
|
) -> MailDomain | None:
|
|
if not company_domain_name:
|
|
return None
|
|
return await session.scalar(
|
|
select(MailDomain)
|
|
.options(selectinload(MailDomain.dns_records))
|
|
.where(
|
|
MailDomain.domain_name == company_domain_name,
|
|
MailDomain.account_id == account_id,
|
|
)
|
|
)
|
|
|
|
|
|
async def create_smtp_credential(
|
|
session: AsyncSession,
|
|
domain: MailDomain,
|
|
*,
|
|
label: str,
|
|
) -> tuple[SmtpCredential, str]:
|
|
username = generate_smtp_username(domain.domain_name)
|
|
plain_password = generate_smtp_password()
|
|
credential = SmtpCredential(
|
|
domain_id=domain.id,
|
|
username=username,
|
|
password_hash=password_hasher.hash(plain_password),
|
|
label=label,
|
|
is_active=True,
|
|
)
|
|
session.add(credential)
|
|
await session.flush()
|
|
return credential, plain_password
|
|
|
|
|
|
async def ensure_platform_domain(session: AsyncSession) -> MailDomain:
|
|
settings = get_settings()
|
|
domain = await get_domain_by_name(session, settings.platform_domain)
|
|
if domain:
|
|
return domain
|
|
return await provision_mail_domain(
|
|
session,
|
|
domain_name=settings.platform_domain,
|
|
domain_type="platform",
|
|
)
|