fix(security): scope owner-less email accounts to a mailbox match in route guards (#5238)

The HTTP email route guard `_assert_owns_account` and the explicit-account_id
path in `_get_email_config` gated cross-tenant access with
`if row.owner and row.owner != owner` -- which skips the check entirely when the
account row is owner-less (owner NULL or ""). `email_accounts` is the one
owner-scoped table left out of the legacy-owner migration backfill
(core/database.py), so such rows persist on multi-user deploys: an account
configured while auth was disabled, or an imported legacy row. Any authenticated
user could then pass that account's id to read/send/update-credentials/delete
another tenant's mailbox and read its decrypted IMAP/SMTP creds.

Both sibling paths already enforce the intended contract -- the same-file
`_owner_or_matching_legacy_account` fallback and the MCP `_account_visible_to_owner`
gate (whose comment says it mirrors "the HTTP email route fallback") only expose
an owner-less account when its own mailbox (imap_user / from_address) is the
caller's. Factor that row-level predicate into `_account_visible_to_owner` and
use it in both guards, so owner-less accounts are visible only on a mailbox
match. Owned accounts, the legacy-claim path, and single-user mode (owner == "")
are unchanged.

Complements #5234 (which fixes the same class on the MCP tool layer); this is
the HTTP route layer it does not touch.

Author: ashvinctrl

routes/email_helpers.py

@@ -349,7 +349,7 @@ def _assert_owns_account(account_id: str, owner: str) -> None:
             row = db.query(_EA).filter(_EA.id == account_id).first()
             if row is None:
                 raise HTTPException(404, "Account not found")
-            if row.owner and row.owner != owner:
+            if not _account_visible_to_owner(row, owner):
                 # Treat as 404 (not 403) so we don't leak existence.
                 raise HTTPException(404, "Account not found")
         finally:
@@ -362,6 +362,26 @@ def _assert_owns_account(account_id: str, owner: str) -> None:
         logger.error(f"Account-owner check failed: {e}")
         raise HTTPException(503, "Account check failed")
 
+
+def _account_visible_to_owner(row, owner: str) -> bool:
+    """Whether an authenticated `owner` may act on this EmailAccount row.
+
+    Mirrors the SQL predicate in `_get_email_config`'s
+    `_owner_or_matching_legacy_account`: a caller sees an account they own, or a
+    legacy owner-less account (owner NULL/"") only when its own mailbox
+    (`imap_user` / `from_address`) is the caller's. `email_accounts` is the one
+    owner-scoped table deliberately left out of the legacy-owner migration
+    backfill, so ownerless rows persist on multi-user deploys — making this the
+    gate that keeps one tenant off another's imported mailbox and its decrypted
+    IMAP/SMTP credentials."""
+    row_owner = getattr(row, "owner", None) or ""
+    if row_owner:
+        return row_owner == owner
+    return owner in {
+        getattr(row, "imap_user", None) or "",
+        getattr(row, "from_address", None) or "",
+    }
+
 def _q(name: str) -> str:
     """Quote an IMAP mailbox name. Defensive: escapes `\\` and `"` and wraps
     in double quotes so user-supplied folder names with spaces or quotes can't
@@ -903,12 +923,13 @@ def _owner_or_matching_legacy_account(query):
         try:
             if account_id:
                 row = db.query(_EA).filter(_EA.id == account_id, _EA.enabled == True).first()  # noqa: E712
-                # If the resolved row belongs to a different owner, treat as
+                # If the resolved row isn't visible to this owner, treat as
                 # not-found rather than silently serving it. This is a defense
                 # in depth — `require_owner` already calls `_assert_owns_account`
                 # for query-param account_ids, but other callers (cookbook
-                # rules, scheduled poller) may not.
-                if row is not None and owner and row.owner and row.owner != owner:
+                # rules, scheduled poller) may not. Ownerless legacy rows are
+                # only visible on a mailbox match, same as the fallback below.
+                if row is not None and owner and not _account_visible_to_owner(row, owner):
                     row = None
             # Fallback path — restrict to this owner's accounts so we don't
             # leak another user's default mailbox to an unconfigured user.

tests/test_email_ownerless_account_owner_scope.py

@@ -0,0 +1,117 @@
+"""Cross-tenant access control for legacy owner-less email accounts.
+
+`email_accounts` is the one owner-scoped table left out of the legacy-owner
+migration backfill (core/database.py), so rows with owner NULL/"" persist on a
+multi-user deploy — e.g. an account configured while auth was disabled, or an
+imported legacy row. The HTTP route guards (`_assert_owns_account` and the
+explicit-account_id path in `_get_email_config`) must scope such rows to a
+mailbox match, exactly like the `_owner_or_matching_legacy_account` fallback and
+the MCP `_account_visible_to_owner` gate. Otherwise any authenticated user can
+read/send/update-credentials/delete another tenant's imported mailbox.
+"""
+
+from unittest import mock
+
+import pytest
+from fastapi import HTTPException
+
+
+def _make_db():
+    from sqlalchemy import create_engine
+    from sqlalchemy.orm import sessionmaker
+    from core.database import Base
+    engine = create_engine("sqlite:///:memory:", connect_args={"check_same_thread": False})
+    Base.metadata.create_all(engine)
+    Factory = sessionmaker(bind=engine)
+    return Factory
+
+
+def _make_account(Factory, account_id, owner, imap_user, from_address="", is_default=False):
+    from core.database import EmailAccount
+    db = Factory()
+    row = EmailAccount(
+        id=account_id,
+        owner=owner,
+        name="Test",
+        enabled=True,
+        is_default=is_default,
+        imap_host="imap.example.com",
+        imap_port=993,
+        imap_user=imap_user,
+        smtp_host="smtp.example.com",
+        smtp_port=587,
+        smtp_user=imap_user,
+        from_address=from_address or imap_user,
+    )
+    db.add(row)
+    db.commit()
+    db.close()
+
+
+def test_assert_owns_account_rejects_ownerless_account_for_other_tenant():
+    """The core regression: a legacy owner-less mailbox is NOT accessible to an
+    authenticated caller whose own mailbox does not match it."""
+    from routes.email_helpers import _assert_owns_account
+    Factory = _make_db()
+    # owner="" (created while auth was disabled); mailbox belongs to victim.
+    _make_account(Factory, "acct-legacy", owner="", imap_user="victim@corp.com")
+
+    with mock.patch("core.database.SessionLocal", Factory):
+        with pytest.raises(HTTPException) as exc:
+            _assert_owns_account("acct-legacy", "attacker")
+    assert exc.value.status_code == 404
+
+
+def test_assert_owns_account_allows_owned_account():
+    from routes.email_helpers import _assert_owns_account
+    Factory = _make_db()
+    _make_account(Factory, "acct-bob", owner="bob", imap_user="bob@corp.com")
+    with mock.patch("core.database.SessionLocal", Factory):
+        _assert_owns_account("acct-bob", "bob")  # no raise
+
+
+def test_assert_owns_account_allows_ownerless_account_on_mailbox_match():
+    """Legacy-claim path stays intact: the user whose mailbox matches an
+    owner-less account may still act on it (imap_user or from_address)."""
+    from routes.email_helpers import _assert_owns_account
+    Factory = _make_db()
+    _make_account(Factory, "acct-legacy", owner="", imap_user="alice@corp.com")
+    with mock.patch("core.database.SessionLocal", Factory):
+        _assert_owns_account("acct-legacy", "alice@corp.com")  # no raise
+
+
+def test_assert_owns_account_noop_for_single_user_mode():
+    """owner == "" (unconfigured / single-user) accepts any account, unchanged."""
+    from routes.email_helpers import _assert_owns_account
+    Factory = _make_db()
+    _make_account(Factory, "acct-legacy", owner="", imap_user="whoever@corp.com")
+    with mock.patch("core.database.SessionLocal", Factory):
+        _assert_owns_account("acct-legacy", "")  # no raise
+
+
+def test_get_email_config_does_not_resolve_ownerless_account_for_other_tenant(monkeypatch):
+    """`_get_email_config(account_id=..., owner=...)` must not serve an
+    owner-less account (and its decrypted creds) to a non-matching tenant."""
+    import routes.email_helpers as eh
+    Factory = _make_db()
+    _make_account(Factory, "acct-legacy", owner="", imap_user="victim@corp.com", is_default=True)
+
+    # Make the settings.json / env fallback empty and deterministic.
+    monkeypatch.setattr(eh, "_load_settings", lambda: {}, raising=False)
+    for var in ("IMAP_HOST", "SMTP_HOST", "IMAP_USER", "SMTP_USER"):
+        monkeypatch.delenv(var, raising=False)
+
+    with mock.patch("core.database.SessionLocal", Factory):
+        cfg = eh._get_email_config(account_id="acct-legacy", owner="attacker")
+
+    assert cfg.get("account_id") != "acct-legacy"
+
+
+def test_get_email_config_resolves_ownerless_account_on_mailbox_match():
+    """The mailbox owner still resolves their claimable legacy account by id."""
+    import routes.email_helpers as eh
+    Factory = _make_db()
+    _make_account(Factory, "acct-legacy", owner="", imap_user="alice@corp.com")
+    with mock.patch("core.database.SessionLocal", Factory):
+        cfg = eh._get_email_config(account_id="acct-legacy", owner="alice@corp.com")
+    assert cfg.get("account_id") == "acct-legacy"