Blame for backend/mailboxes.py

"""
Mailbox connections — the relay identity an org sends through (docs/blueprint/adr/0001).
Two kinds, both stored the same way once connected: "oauth" (Gmail/Outlook/Yahoo — see
backend/core/constants.py SUPPORTED_OAUTH_PROVIDERS) and "smtp" (hosting-provider mailboxes,
raw host/port/username/password). Credentials are encrypted at rest via backend.crypto_utils
(the prototype stand-in for ADR-0002's GCP KMS envelope encryption).

Verification is the 3-OTP round-trip from SRS FR-6: send unique OTPs from the connected
mailbox to M8ion's own test mailbox, then confirm receipt/content/metadata match before the
connection is usable for real sends. The five failure states below are SRS FR-7, verbatim.
"""

from __future__ import annotations

import imaplib
import json
import re
import uuid
from datetime import datetime, timedelta, timezone

from backend import delivery
from backend import firestore_client as db
from backend import mailer_client
from backend import orgs
from backend.core.constants import (
    MAILBOX_CREDENTIAL_CHANGE_DAYS,
    MAILBOX_DAILY_LIMIT_MAX,
    SYSTEM_SENDER_ADDRESS_PROTO,
    SYSTEM_TEST_MAILBOX_ADDRESS_PROTO,
    SUPPORTED_OAUTH_PROVIDERS,
    VERIFICATION_OTP_COUNT,
)
from backend.crypto_utils import decrypt, encrypt
from backend.mock_mailer import AuthError, ConnectionFailure, MockInbox, MockMailer

# Stands in for "M8ion's own test mailbox" until a real domain/inbox exists — used for the
# mock round-trip (REAL_SEND_ENABLED=false). The real round-trip (true) has no inbox to
# read back yet (that needs IMAP, unbuilt — see _verify_real's docstring), so it sends to
# SYSTEM_SENDER_ADDRESS_PROTO instead, the one real mailbox this project actually holds.
SYSTEM_TEST_INBOX = MockInbox()

# SRS FR-7, worded exactly as specified.
FAILURE_MESSAGES = {
    "auth_rejected": "We couldn't sign in to that mailbox. Double-check the email and password, then try again.",
    "connection_failed": "We couldn't reach that mail server. Check the server address and port, or confirm with your email provider that SMTP access is enabled.",
    "silent_failure": "We couldn't verify delivery. Contact support and we'll help you troubleshoot.",
    "unreliable": "This mailbox is only delivering some of the time, which usually means intermittent throttling from your provider. We recommend resolving this before relying on it for real sends.",
    "content_mismatch": "Something didn't match during verification. This connection has been flagged for manual review.",
    "imap_failed": "Sending works, but we couldn't sign in to the incoming (IMAP) side. Check the IMAP server, port, username and password. We need it to show you delays, bounces and replies.",
}

# Honest scope for the real (non-mocked) path: auth + send are genuinely verified;
# arrival isn't, since reading a real inbox back needs IMAP (Phase 2, unbuilt).
VERIFIED_REAL_MESSAGE = (
    "Your credentials were accepted and a test message was sent through your mail server. "
    "We can't yet confirm it arrived — check the inbox yourself for full certainty."
)


SEND_FAILED_MESSAGE = (
    "We signed in to that mailbox, but your provider refused the test message. "
    "Check that sending is allowed for this account, then try again."
)


class ValidationError(Exception):
    pass


def _now() -> str:
    return datetime.now(timezone.utc).isoformat()


def _connection_path(org_id: str, connection_id: str) -> str:
    return f"orgs/{org_id}/mailboxes/{connection_id}"


def connect_mailbox(org_id: str, actor_uid: str, kind: str, credentials: dict) -> str:
    """Registers a new, not-yet-verified mailbox connection. An org can have more than one
    (SRS FR-5). Requires admin or owner (SRS: mailbox management is an org-management
    action, not a day-to-day sending one)."""
    orgs.require_permission(org_id, actor_uid, "manage_mailboxes")

    if kind == "oauth":
        provider = credentials.get("provider")
        if provider not in SUPPORTED_OAUTH_PROVIDERS:
            raise ValidationError(f"Unsupported OAuth provider {provider!r}")
    elif kind == "smtp":
        required = {"host", "port", "username", "password"}
        missing = required - credentials.keys()
        if missing:
            raise ValidationError(f"SMTP connection missing fields: {missing}")
        credentials = _clean_imap(credentials, required=True)
        # Spaces in an app password are display grouping, never part of it.
        credentials = {**credentials, "password": "".join(str(credentials["password"]).split())}
    else:
        raise ValidationError(f"Unknown mailbox kind {kind!r}")

    # The sending address isn't a secret — it's the From: on every message this connection
    # sends — so it's kept plain alongside the encrypted blob rather than requiring a
    # decrypt just to list connections (SRS FR-5's list view needs this on every load).
    email = credentials.get("username") if kind == "smtp" else credentials.get("email")

    connection_id = uuid.uuid4().hex
    db.set_doc(
        _connection_path(org_id, connection_id),
        {
            "kind": kind,
            "email": email,
            "has_imap": bool(credentials.get("imap_host")),
            "encrypted_credentials": encrypt(json.dumps(credentials)),
            "status": "pending_verification",
            "status_message": None,
            "created_at": _now(),
            "verified_at": None,
        },
    )
    return connection_id


def _clean_imap(credentials: dict, required: bool) -> dict:
    """Checks the incoming (IMAP) half of an SMTP mailbox. Both halves are needed: SMTP sends,
    IMAP reads back what the provider did with the message (delays, bounces, replies). The IMAP
    user and password may be left out when they are the same as the SMTP ones."""
    host = str(credentials.get("imap_host") or "").strip()
    if not host:
        if required:
            raise ValidationError("Add the incoming (IMAP) server too. M8ion needs it to show delays, bounces and replies.")
        return credentials
    try:
        port = int(credentials.get("imap_port") or 993)
    except (TypeError, ValueError) as exc:
        raise ValidationError("The IMAP port must be a number, usually 993") from exc
    if not 1 <= port <= 65535:
        raise ValidationError("The IMAP port must be a number, usually 993")
    cleaned = {**credentials, "imap_host": host, "imap_port": port}
    for key in ("imap_username", "imap_password"):
        if not str(cleaned.get(key) or "").strip():
            cleaned.pop(key, None)
    if cleaned.get("imap_password"):
        cleaned["imap_password"] = "".join(str(cleaned["imap_password"]).split())
    return cleaned


def get_mailbox(org_id: str, connection_id: str) -> dict:
    doc = db.get_doc(_connection_path(org_id, connection_id))
    if doc is None:
        raise orgs.NotFoundError(f"No mailbox connection {connection_id!r} for org {org_id!r}")
    return doc


def list_mailboxes(org_id: str) -> dict[str, dict]:
    prefix = f"orgs/{org_id}/mailboxes/"
    docs = db.query_prefix(prefix)
    return {path[len(prefix):]: data for path, data in docs.items()}


def set_dismissed(org_id: str, actor_uid: str, connection_id: str, dismissed: bool) -> None:
    """Hides (or restores) a failed try from the Mailboxes page and the attention counts. The
    connection itself is kept, so this is fully reversible."""
    orgs.require_permission(org_id, actor_uid, "manage_mailboxes")
    connection = get_mailbox(org_id, connection_id)
    if dismissed and connection.get("status") in ("verified", "pending_verification"):
        raise ValidationError("Only a mailbox that did not connect can be dismissed")
    db.update_doc(_connection_path(org_id, connection_id), {"dismissed": dismissed})


# Per-mailbox choices. Empty / None means "use the organisation's setting".
DEFAULT_MAILBOX_SETTINGS = {"label": "", "from_name": "", "reply_to": "", "daily_limit": 0, "save_sent_copy": None}
_EMAIL = re.compile(r"^[^@\s]+@[^@\s]+\.[^@\s]+$")


def mailbox_settings(connection: dict) -> dict:
    return DEFAULT_MAILBOX_SETTINGS | {k: v for k, v in (connection.get("settings") or {}).items() if k in DEFAULT_MAILBOX_SETTINGS}


def update_settings(org_id: str, actor_uid: str, connection_id: str, values: dict) -> dict:
    """Saves this mailbox's own sender name, reply-to, daily limit, Sent-copy choice and label."""
    orgs.require_permission(org_id, actor_uid, "manage_mailboxes")
    connection = get_mailbox(org_id, connection_id)
    unknown = set(values) - set(DEFAULT_MAILBOX_SETTINGS)
    if unknown:
        raise ValidationError(f"Unknown setting: {', '.join(sorted(unknown))}")
    merged = mailbox_settings(connection) | values
    for key in ("label", "from_name"):
        merged[key] = " ".join(str(merged[key] or "").split())
        if len(merged[key]) > 80:
            raise ValidationError("Keep names to 80 characters or fewer")
    merged["reply_to"] = str(merged["reply_to"] or "").strip()
    if merged["reply_to"] and not _EMAIL.match(merged["reply_to"]):
        raise ValidationError("Replies address doesn't look like an email address")
    try:
        merged["daily_limit"] = int(merged["daily_limit"] or 0)
    except (TypeError, ValueError) as exc:
        raise ValidationError("Daily limit must be a whole number") from exc
    if not 0 <= merged["daily_limit"] <= MAILBOX_DAILY_LIMIT_MAX:
        raise ValidationError(f"Daily limit must be between 0 (no limit) and {MAILBOX_DAILY_LIMIT_MAX:,}")
    if merged["save_sent_copy"] is not None:
        merged["save_sent_copy"] = bool(merged["save_sent_copy"])
    db.update_doc(_connection_path(org_id, connection_id), {"settings": merged})
    return merged


def next_credential_change(connection: dict) -> str | None:
    """When the sign-in details may next be changed, or None if they may be changed now."""
    last = connection.get("credentials_changed_at")
    if not last:
        return None
    allowed = datetime.fromisoformat(last) + timedelta(days=MAILBOX_CREDENTIAL_CHANGE_DAYS)
    return allowed.isoformat() if allowed > datetime.now(timezone.utc) else None


def change_credentials(org_id: str, actor_uid: str, connection_id: str, credentials: dict) -> dict:
    """Replaces an SMTP mailbox's server and sign-in details, at most once every
    MAILBOX_CREDENTIAL_CHANGE_DAYS. The new details are tested first; if the test fails
    the old ones stay in place and the change doesn't count. Returns
    {status, status_message, changed}."""
    orgs.require_permission(org_id, actor_uid, "manage_mailboxes")
    connection = get_mailbox(org_id, connection_id)
    if connection["kind"] != "smtp":
        raise ValidationError("Only mailboxes connected with server details can be changed here")
    wait = next_credential_change(connection)
    # Adding the incoming (IMAP) side to a mailbox that has none is allowed at any time: it only
    # turns on delivery tracking and does not touch how the mailbox signs in to send.
    adding_imap = not connection.get("has_imap") and bool(credentials) and all(k.startswith("imap_") for k in credentials)
    if wait and not adding_imap:
        raise ValidationError(f"These details were changed recently. They can be changed again from {wait[:10]}.")
    current = _decrypted_credentials(connection)
    merged = {**current, **{k: v for k, v in credentials.items() if v not in (None, "")}}
    missing = {"host", "port", "username", "password"} - merged.keys()
    if missing:
        raise ValidationError(f"SMTP connection missing fields: {missing}")
    merged["password"] = "".join(str(merged["password"]).split())
    merged = _clean_imap(merged, required=False)
    if merged == current:
        raise ValidationError("Nothing changed")

    path = _connection_path(org_id, connection_id)
    if adding_imap:
        # Only the incoming side is new: test just that. Re-sending the test emails would be slow
        # and is not needed, since sending is unchanged.
        if mailer_client.real_send_enabled():
            details = delivery.imap_details(merged)
            try:
                delivery.open_inbox(details).close()
            except (imaplib.IMAP4.error, OSError):
                return {"status": "imap_failed", "status_message": FAILURE_MESSAGES["imap_failed"], "changed": False}
        db.update_doc(path, {"has_imap": True, "encrypted_credentials": encrypt(json.dumps(merged))})
        return {"status": connection.get("status"), "status_message": connection.get("status_message"), "changed": True}
    keep = {k: connection.get(k) for k in ("email", "has_imap", "encrypted_credentials", "status", "status_message", "verified_at")}
    db.update_doc(path, {"email": merged["username"], "has_imap": bool(merged.get("imap_host")), "encrypted_credentials": encrypt(json.dumps(merged)), "status": "pending_verification"})
    status = verify_mailbox(org_id, connection_id)
    if status != "verified":
        message = get_mailbox(org_id, connection_id).get("status_message")
        db.update_doc(path, keep)
        return {"status": status, "status_message": message, "changed": False}
    if not adding_imap:
        db.update_doc(path, {"credentials_changed_at": _now()})
    return {"status": status, "status_message": get_mailbox(org_id, connection_id).get("status_message"), "changed": True}


def remove_mailbox(org_id: str, actor_uid: str, connection_id: str) -> None:
    """Deletes the connection and its stored sign-in details. Past sends keep their record.
    Refused while a sign-up form sends its confirmation emails through it."""
    orgs.require_permission(org_id, actor_uid, "manage_mailboxes")
    get_mailbox(org_id, connection_id)
    users = [
        data.get("name") or "a sign-up form"
        for data in db.query_prefix(f"orgs/{org_id}/collect_keys/").values()
        if not data.get("revoked") and data.get("mailbox_connection_id") == connection_id
    ]
    if users:
        raise ValidationError(f"{users[0]} sends its confirmation emails from this mailbox. Pick another mailbox there first.")
    db.delete_doc(_connection_path(org_id, connection_id))
    for path, brand in db.query_prefix(f"orgs/{org_id}/brands/").items():
        if connection_id in brand.get("mailbox_ids", []):
            db.update_doc(path, {"mailbox_ids": [i for i in brand["mailbox_ids"] if i != connection_id]})
    emails_path = f"orgs/{org_id}/settings/emails"
    prefs = db.get_doc(emails_path)
    if prefs and prefs.get("default_mailbox_id") == connection_id:
        db.set_doc(emails_path, {**prefs, "default_mailbox_id": ""})


def get_decrypted_credentials(org_id: str, connection_id: str) -> dict:
    """The only sanctioned way to read a connection's plaintext credentials — callers
    outside this module should never touch encrypted_credentials directly."""
    return _decrypted_credentials(get_mailbox(org_id, connection_id))


def _decrypted_credentials(connection: dict) -> dict:
    return json.loads(decrypt(connection["encrypted_credentials"]))


def verify_mailbox(org_id: str, connection_id: str) -> str:
    """Runs the verification round-trip and updates the connection's status. Returns the
    final status string. Never raises for an expected failure mode — those are all
    recorded as a status, per SRS FR-7 ("robust support", not a bare exception).

    Branches on mailer_client.real_send_enabled(): mocked (default) runs the full 5-state
    round-trip against an in-memory inbox; real runs _verify_real, which can only reach
    3 of those 5 states — see its docstring for exactly why."""
    connection = get_mailbox(org_id, connection_id)
    credentials = _decrypted_credentials(connection)

    if mailer_client.real_send_enabled():
        return _verify_real(org_id, connection_id, connection["kind"], credentials)
    return _verify_mock(org_id, connection_id, credentials)


def _verify_mock(org_id: str, connection_id: str, credentials: dict) -> str:
    try:
        mailer = MockMailer(credentials, SYSTEM_TEST_INBOX)
    except AuthError:
        return _settle(org_id, connection_id, "auth_rejected")
    except ConnectionFailure:
        return _settle(org_id, connection_id, "connection_failed")

    otps = [uuid.uuid4().hex[:8] for _ in range(VERIFICATION_OTP_COUNT)]
    for i, otp in enumerate(otps):
        mailer.send(
            to_addr=SYSTEM_TEST_MAILBOX_ADDRESS_PROTO,
            subject=f"M8ion verification {connection_id} #{i}",
            body=f"OTP:{otp}",
        )

    arrived = [
        m for m in SYSTEM_TEST_INBOX.messages_to(SYSTEM_TEST_MAILBOX_ADDRESS_PROTO)
        if connection_id in m.subject
    ]

    if not arrived:
        return _settle(org_id, connection_id, "silent_failure")

    if len(arrived) < len(otps):
        _delete_test_messages(arrived)
        return _settle(org_id, connection_id, "unreliable")

    expected_bodies = {f"OTP:{otp}" for otp in otps}
    arrived_bodies = {m.body for m in arrived}
    if arrived_bodies != expected_bodies:
        _delete_test_messages(arrived)
        return _settle(org_id, connection_id, "content_mismatch")

    _delete_test_messages(arrived)
    return _settle(org_id, connection_id, "verified")


def _verify_real(org_id: str, connection_id: str, kind: str, credentials: dict) -> str:
    """Real SMTP auth + a real send attempt against SYSTEM_SENDER_ADDRESS_PROTO — the one
    real mailbox this project holds credentials for. Can only ever reach "auth_rejected",
    "connection_failed", or "verified" — "silent_failure"/"unreliable"/"content_mismatch"
    all require reading the destination inbox back (IMAP) to detect, which isn't built
    (Phase 2's bounce-detection work). A "verified" here means auth + send succeeded, not
    that arrival was confirmed — VERIFIED_REAL_MESSAGE says so explicitly in the UI."""
    try:
        mailer = mailer_client.get_mailer(kind, credentials)
    except AuthError:
        return _settle(org_id, connection_id, "auth_rejected")
    except (ConnectionFailure, NotImplementedError):
        return _settle(org_id, connection_id, "connection_failed")

    try:
        for i in range(VERIFICATION_OTP_COUNT):
            otp = uuid.uuid4().hex[:8]
            mailer.send(
                to_addr=SYSTEM_SENDER_ADDRESS_PROTO,
                subject=f"M8ion verification {connection_id} #{i}",
                body=f"OTP:{otp}",
            )
    except (AuthError, ConnectionFailure):
        # Sign-in already worked, so telling the person the server was unreachable would be wrong.
        return _settle(org_id, connection_id, "connection_failed", message_override=SEND_FAILED_MESSAGE)

    details = delivery.imap_details(credentials)
    if details is not None:
        try:
            delivery.open_inbox(details).close()
        except (imaplib.IMAP4.error, OSError):
            return _settle(org_id, connection_id, "imap_failed")
    return _settle(org_id, connection_id, "verified", message_override=VERIFIED_REAL_MESSAGE)


def _delete_test_messages(messages) -> None:
    """The test mailbox doesn't keep verification OTPs around once a round-trip is
    evaluated, matching the "delete the test mail after test complete" requirement."""
    for m in messages:
        if m in SYSTEM_TEST_INBOX.messages:
            SYSTEM_TEST_INBOX.messages.remove(m)


def _settle(org_id: str, connection_id: str, status: str, message_override: str | None = None) -> str:
    path = _connection_path(org_id, connection_id)
    db.update_doc(
        path,
        {
            "status": status,
            "status_message": message_override if message_override is not None else FAILURE_MESSAGES.get(status),
            "verified_at": _now() if status == "verified" else None,
            "dismissed": False,  # a fresh result is news again, even if the last failure was dismissed
        },
    )
    return status