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.
This commit is contained in:
Ashvin
2026-07-05 17:20:32 +05:30
committed by GitHub
parent 5acd0ceae9
commit c8169ad7a9
2 changed files with 142 additions and 4 deletions
+25 -4
View File
@@ -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 _get_email_config(account_id: str | None = None, owner: str = "") -> dict:
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.
@@ -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"