first commit

backend/app/services/__init__.py

@@ -0,0 +1 @@
+"""Backend service modules."""

backend/app/services/dns_verification.py

@@ -0,0 +1,176 @@
+"""Live DNS checks for company domain onboarding."""
+
+from __future__ import annotations
+
+import dns.exception
+import dns.resolver
+
+
+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 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 getall.help 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 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."
+
+
+def verify_company_dns(
+    *,
+    domain_name: str,
+    shared_mail_host: str,
+    dkim_selector: str,
+    dkim_public_key: str,
+    platform_spf_include: str,
+) -> list[dict]:
+    dmarc_expected = f"v=DMARC1; p=quarantine; rua=mailto:dmarc@{domain_name}"
+    checks = [
+        ("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),
+        ),
+        ("dmarc", *verify_dmarc(domain_name, dmarc_expected)),
+    ]
+    return [
+        {
+            "purpose": purpose,
+            "verified": verified,
+            "message": message,
+        }
+        for purpose, verified, message in checks
+    ]

backend/app/services/domain_provisioning.py

@@ -0,0 +1,380 @@
+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,
+) -> list[dict]:
+    """Zoho-style records: shared MX host, SPF include, per-domain DKIM/DMARC."""
+    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": "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
+        dns_records = build_company_dns_records(
+            normalized,
+            settings.smtp_public_host,
+            dkim_selector,
+            public_key,
+            settings.platform_spf_include,
+        )
+    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 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",
+    )

backend/app/services/inbound_smtp.py

@@ -0,0 +1,159 @@
+import asyncio
+import ipaddress
+import logging
+from datetime import datetime, timezone
+
+from aiosmtpd.smtp import SMTP
+from sqlalchemy import func, select
+
+from app.config import get_settings
+from app.database import SessionFactory, mongo_db
+from app.models import Account
+from app.services.mime_parser import parse_mime
+from app.services.raw_mime_storage import delivery_digest, get_raw_mime_storage
+
+logger = logging.getLogger(__name__)
+
+
+def peer_is_allowed(session) -> bool:
+    if session is None or not getattr(session, "peer", None):
+        return True
+    try:
+        peer = ipaddress.ip_address(session.peer[0])
+        return any(
+            peer in ipaddress.ip_network(cidr)
+            for cidr in get_settings().inbound_allowed_cidrs
+        )
+    except ValueError:
+        return False
+
+
+class InboundHandler:
+    async def handle_RCPT(self, server, session, envelope, address, rcpt_options):
+        if not peer_is_allowed(session):
+            logger.warning("Rejected inbound SMTP peer %s", session.peer[0])
+            return "550 5.7.1 Delivery from this host is not authorized"
+        normalized = address.strip().lower()
+        async with SessionFactory() as database_session:
+            account = await database_session.scalar(
+                select(Account).where(func.lower(Account.email) == normalized)
+            )
+        if account is None:
+            return "550 5.1.1 Recipient address rejected: user unknown"
+        envelope.rcpt_tos.append(normalized)
+        envelope.rcpt_options.extend(rcpt_options)
+        if not hasattr(envelope, "account_ids"):
+            envelope.account_ids = {}
+        envelope.account_ids[normalized] = str(account.id)
+        return "250 2.1.5 Recipient OK"
+
+    async def handle_DATA(self, server, session, envelope):
+        settings = get_settings()
+        raw = envelope.original_content
+        if len(raw) > settings.inbound_max_bytes:
+            return "552 5.3.4 Message size exceeds fixed maximum message size"
+        recipients = list(dict.fromkeys(address.lower() for address in envelope.rcpt_tos))
+        if not recipients:
+            return "554 5.5.1 No valid recipients"
+
+        delivery_key = delivery_digest(raw, recipients)
+        try:
+            tombstones = getattr(mongo_db, "message_tombstones", None)
+            if tombstones is not None and await tombstones.find_one(
+                {"delivery_key": delivery_key},
+                {"_id": 1},
+            ):
+                return f"250 2.0.0 Message previously accepted as {delivery_key}"
+            parsed = parse_mime(raw)
+            body_text = parsed.pop("body_text")
+            body_html = parsed.pop("body_html")
+            stored = await get_raw_mime_storage().put(raw, delivery_key)
+            now = datetime.now(timezone.utc)
+            message = {
+                **parsed,
+                "direction": "inbound",
+                "envelope_sender": envelope.mail_from.lower(),
+                "envelope_recipients": recipients,
+                "delivery_key": delivery_key,
+                "raw_mime": {
+                    "bucket": stored.bucket,
+                    "name": stored.name,
+                    "generation": stored.generation,
+                    "size": stored.size,
+                    "crc32c": stored.crc32c,
+                },
+                "received_at": now,
+                "created_at": now,
+                "has_text_body": bool(body_text),
+                "has_html_body": bool(body_html),
+            }
+            await mongo_db.messages.update_one(
+                {"delivery_key": delivery_key},
+                {"$setOnInsert": message},
+                upsert=True,
+            )
+            persisted = await mongo_db.messages.find_one(
+                {"delivery_key": delivery_key},
+                {"_id": 1},
+            )
+            if persisted is None:
+                raise RuntimeError("Inbound message was not persisted.")
+            for recipient in recipients:
+                account_id = envelope.account_ids[recipient]
+                await mongo_db.mailbox_entries.update_one(
+                    {
+                        "account_id": account_id,
+                        "message_id": persisted["_id"],
+                        "mailbox": "inbox",
+                    },
+                    {
+                        "$setOnInsert": {
+                            "account_id": account_id,
+                            "message_id": persisted["_id"],
+                            "mailbox": "inbox",
+                            "direction": "inbound",
+                            "sender": parsed["sender"],
+                            "recipients": parsed["to"],
+                            "subject": parsed["subject"],
+                            "snippet": parsed["snippet"],
+                            "attachments": parsed["attachments"],
+                            "status": "received",
+                            "is_read": False,
+                            "is_starred": False,
+                            "event_at": now,
+                            "created_at": now,
+                            "trashed_at": None,
+                            "deletion_state": None,
+                        }
+                    },
+                    upsert=True,
+                )
+        except Exception:
+            logger.exception("Temporary failure accepting inbound message")
+            return "451 4.3.0 Temporary server failure; please retry"
+        return f"250 2.0.0 Message accepted as {delivery_key}"
+
+
+class InboundSMTPServer:
+    def __init__(self):
+        self.server: asyncio.Server | None = None
+
+    async def start(self) -> None:
+        settings = get_settings()
+        loop = asyncio.get_running_loop()
+        self.server = await loop.create_server(
+            lambda: SMTP(
+                InboundHandler(),
+                data_size_limit=settings.inbound_max_bytes,
+                enable_SMTPUTF8=True,
+                decode_data=False,
+            ),
+            host=settings.inbound_smtp_host,
+            port=settings.inbound_smtp_port,
+        )
+
+    async def stop(self) -> None:
+        if self.server is not None:
+            self.server.close()
+            await self.server.wait_closed()
+            self.server = None

backend/app/services/mail_sender.py

@@ -0,0 +1,107 @@
+import email.policy
+import email.utils
+from email.message import EmailMessage
+
+import aiosmtplib
+import dkim
+
+from app.config import get_settings
+from app.models import MailDomain
+
+SMTP_POLICY = email.policy.SMTP.clone(max_line_length=998)
+
+
+def build_message(
+    *,
+    sender: str,
+    recipients: list[str],
+    cc: list[str],
+    subject: str,
+    body_text: str,
+    body_html: str | None,
+) -> EmailMessage:
+    message = EmailMessage(policy=SMTP_POLICY)
+    message["From"] = sender
+    message["To"] = ", ".join(recipients)
+    if cc:
+        message["Cc"] = ", ".join(cc)
+    message["Subject"] = subject
+    message["Date"] = email.utils.formatdate(localtime=True)
+    message["Message-ID"] = email.utils.make_msgid(domain=sender.split("@")[-1])
+    message.set_content(body_text)
+    if body_html:
+        message.add_alternative(body_html, subtype="html")
+    return message
+
+
+def sign_message(message: EmailMessage, domain: MailDomain) -> bytes:
+    raw_message = message.as_bytes(policy=SMTP_POLICY)
+    signed_headers = [
+        b"from",
+        b"to",
+        b"subject",
+        b"date",
+        b"message-id",
+        b"mime-version",
+        b"content-type",
+    ]
+    if message.get("Cc"):
+        signed_headers.insert(3, b"cc")
+    signature = dkim.sign(
+        raw_message,
+        domain.dkim_selector.encode(),
+        domain.domain_name.encode(),
+        domain.dkim_private_key.encode(),
+        include_headers=signed_headers,
+        canonicalize=(b"relaxed", b"relaxed"),
+    )
+    if not signature.endswith(b"\r\n"):
+        signature += b"\r\n"
+    return signature + raw_message
+
+
+async def send_via_smtp(
+    message_bytes: bytes,
+    *,
+    sender: str,
+    recipients: list[str],
+) -> None:
+    settings = get_settings()
+    await aiosmtplib.send(
+        message_bytes,
+        sender=sender,
+        recipients=recipients,
+        hostname=settings.smtp_host,
+        port=settings.smtp_port,
+        username=settings.smtp_username or None,
+        password=settings.smtp_password or None,
+        start_tls=settings.smtp_use_tls,
+        timeout=settings.smtp_timeout,
+    )
+
+
+async def deliver_email(
+    *,
+    sender: str,
+    recipients: list[str],
+    cc: list[str],
+    bcc: list[str],
+    subject: str,
+    body_text: str,
+    body_html: str | None,
+    domain: MailDomain,
+) -> None:
+    message = build_message(
+        sender=sender,
+        recipients=recipients,
+        cc=cc,
+        subject=subject,
+        body_text=body_text,
+        body_html=body_html,
+    )
+    signed = sign_message(message, domain)
+    await send_via_smtp(
+        signed,
+        sender=sender,
+        recipients=[*recipients, *cc, *bcc],
+    )

backend/app/services/mime_parser.py

@@ -0,0 +1,90 @@
+import re
+from datetime import timezone
+from email import policy
+from email.parser import BytesParser
+from email.utils import getaddresses, parsedate_to_datetime
+from html import unescape
+
+
+def _addresses(values: list[str]) -> list[dict[str, str]]:
+    return [
+        {
+            "name": name.strip() or address.split("@", 1)[0],
+            "address": address.lower(),
+        }
+        for name, address in getaddresses(values)
+        if address
+    ]
+
+
+def _text(part) -> str:
+    try:
+        value = part.get_content()
+    except (LookupError, UnicodeError):
+        payload = part.get_payload(decode=True) or b""
+        value = payload.decode("utf-8", errors="replace")
+    return value if isinstance(value, str) else ""
+
+
+def _snippet(text: str, html: str | None, limit: int = 240) -> str:
+    source = text
+    if not source and html:
+        source = re.sub(r"<[^>]+>", " ", html)
+        source = unescape(source)
+    return re.sub(r"\s+", " ", source).strip()[:limit]
+
+
+def parse_mime(raw: bytes) -> dict:
+    message = BytesParser(policy=policy.default).parsebytes(raw)
+    sender = _addresses(message.get_all("from", []))
+    from_value = sender[0] if sender else {"name": "Unknown sender", "address": ""}
+    text_parts: list[str] = []
+    html_parts: list[str] = []
+    attachments: list[dict] = []
+
+    for part in message.walk():
+        if part.is_multipart():
+            continue
+        disposition = part.get_content_disposition()
+        filename = part.get_filename()
+        content_type = part.get_content_type()
+        payload = part.get_payload(decode=True) or b""
+        if disposition == "attachment" or filename:
+            attachments.append(
+                {
+                    "filename": filename or "attachment",
+                    "content_type": content_type,
+                    "size": len(payload),
+                    "content_id": (part.get("Content-ID") or "").strip("<>") or None,
+                    "disposition": disposition or "attachment",
+                }
+            )
+        elif content_type == "text/plain":
+            text_parts.append(_text(part))
+        elif content_type == "text/html":
+            html_parts.append(_text(part))
+
+    date = None
+    if message.get("Date"):
+        try:
+            date = parsedate_to_datetime(message["Date"])
+            if date.tzinfo is None:
+                date = date.replace(tzinfo=timezone.utc)
+            date = date.astimezone(timezone.utc)
+        except (TypeError, ValueError, OverflowError):
+            pass
+    body_text = "\n".join(text_parts).strip()
+    body_html = "\n".join(html_parts).strip() or None
+    return {
+        "message_id": (message.get("Message-ID") or "").strip() or None,
+        "sender": from_value,
+        "to": _addresses(message.get_all("to", [])),
+        "cc": _addresses(message.get_all("cc", [])),
+        "reply_to": _addresses(message.get_all("reply-to", [])),
+        "subject": str(message.get("Subject") or ""),
+        "sent_at": date,
+        "body_text": body_text,
+        "body_html": body_html,
+        "snippet": _snippet(body_text, body_html),
+        "attachments": attachments,
+    }

backend/app/services/raw_mime_storage.py

@@ -0,0 +1,101 @@
+import asyncio
+import hashlib
+from dataclasses import dataclass
+
+import google_crc32c
+from google.api_core.exceptions import NotFound, PreconditionFailed
+from google.cloud import storage
+
+from app.config import get_settings
+
+
+@dataclass(frozen=True)
+class StoredMime:
+    bucket: str
+    name: str
+    generation: int
+    size: int
+    crc32c: str
+
+
+class RawMimeStorage:
+    def __init__(self, bucket: str, prefix: str = "raw", project: str | None = None):
+        self.bucket_name = bucket
+        self.prefix = prefix.strip("/")
+        self.client = storage.Client(project=project)
+
+    async def put(self, raw: bytes, delivery_key: str) -> StoredMime:
+        return await asyncio.to_thread(self._put, raw, delivery_key)
+
+    def _put(self, raw: bytes, delivery_key: str) -> StoredMime:
+        name = f"{self.prefix}/{delivery_key[:2]}/{delivery_key}.eml"
+        blob = self.client.bucket(self.bucket_name).blob(name)
+        checksum = google_crc32c.Checksum(raw).hexdigest().decode("ascii")
+        try:
+            blob.upload_from_string(
+                raw,
+                content_type="message/rfc822",
+                checksum="crc32c",
+                if_generation_match=0,
+            )
+            blob.reload()
+        except PreconditionFailed:
+            blob.reload()
+        return StoredMime(
+            bucket=self.bucket_name,
+            name=name,
+            generation=int(blob.generation),
+            size=len(raw),
+            crc32c=checksum,
+        )
+
+    async def get(self, reference: dict) -> bytes:
+        return await asyncio.to_thread(self._get, reference)
+
+    def _get(self, reference: dict) -> bytes:
+        generation = int(reference["generation"])
+        blob = self.client.bucket(reference["bucket"]).blob(
+            reference["name"],
+            generation=generation,
+        )
+        return blob.download_as_bytes(if_generation_match=generation)
+
+    async def delete(self, reference: dict) -> None:
+        await asyncio.to_thread(self._delete, reference)
+
+    def _delete(self, reference: dict) -> None:
+        generation = int(reference["generation"])
+        blob = self.client.bucket(reference["bucket"]).blob(
+            reference["name"],
+            generation=generation,
+        )
+        try:
+            blob.delete(if_generation_match=generation)
+        except NotFound:
+            # A prior delete may have removed GCS before Mongo acknowledged it.
+            pass
+
+
+def delivery_digest(raw: bytes, recipients: list[str]) -> str:
+    digest = hashlib.sha256()
+    digest.update(raw)
+    digest.update(b"\0")
+    digest.update("\0".join(sorted(set(recipients))).encode())
+    return digest.hexdigest()
+
+
+_storage: RawMimeStorage | None = None
+
+
+def get_raw_mime_storage() -> RawMimeStorage:
+    global _storage
+    if _storage is None:
+        settings = get_settings()
+        if not settings.gcs_raw_mime_bucket:
+            raise RuntimeError("GCS_RAW_MIME_BUCKET is required for inbound mail.")
+        _storage = RawMimeStorage(
+            settings.gcs_raw_mime_bucket,
+            settings.gcs_raw_mime_prefix,
+            settings.gcs_project,
+        )
+    return _storage