Spaces:
Sleeping
Sleeping
| """ | |
| 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 | |