fix(email): verify mailbox identity on OAuth reconnect (#5648)

routes/email_routes.py

@@ -5170,6 +5170,9 @@ def setup_email_routes():
             return _RR("/?section=integrations&email_oauth_error=token_exchange_failed")
         access_token = data.get("access_token", "")
         refresh_token = data.get("refresh_token", "")
+        if not access_token or not refresh_token:
+            logger.warning("Google token exchange omitted required offline credentials")
+            return _RR("/?section=integrations&email_oauth_error=token_exchange_failed")
         expiry = str(int(time.time()) + data.get("expires_in", 3600))
         # Fetch the email address from userinfo so we can auto-fill imap_user.
         email_addr = ""
@@ -5194,10 +5197,33 @@ def setup_email_routes():
             if owner and row.owner and row.owner != owner:
                 logger.warning("OAuth callback owner mismatch — rejecting token write")
                 return _RR("/?section=integrations&email_oauth_error=ownership_error")
+
+            # A reconnect must prove that the token belongs to the mailbox
+            # already configured on this row. Otherwise authenticating a
+            # different Google account leaves the saved IMAP/SMTP usernames
+            # paired with credentials for another identity.
+            verified_email = (
+                email_addr.strip().casefold()
+                if isinstance(email_addr, str)
+                else ""
+            )
+            configured_logins = {
+                value.strip().casefold()
+                for value in (row.imap_user or "", row.smtp_user or "")
+                if value.strip()
+            }
+            if not verified_email or any(
+                login != verified_email for login in configured_logins
+            ):
+                logger.warning(
+                    "Google OAuth mailbox identity verification failed for account %s",
+                    account_id,
+                )
+                return _RR("/?section=integrations&email_oauth_error=identity_verification_failed")
+
             row.oauth_provider = "google"
             row.oauth_access_token = _enc(access_token)
-            if refresh_token:
+            row.oauth_refresh_token = _enc(refresh_token)
-                row.oauth_refresh_token = _enc(refresh_token)
             row.oauth_token_expiry = expiry
             # Auto-fill Google IMAP/SMTP settings if not already configured.
             if not row.imap_host:

tests/test_email_oauth.py

@@ -372,7 +372,15 @@ async def test_callback_valid_owner_writes_encrypted_tokens_to_intended_account(
     from core.database import EmailAccount
 
     db, Factory = _make_db()
-    _make_account(db, account_id="acct-v", owner="alice", imap_host="", smtp_host="")
+    _make_account(
+        db,
+        account_id="acct-v",
+        owner="alice",
+        imap_host="",
+        smtp_host="",
+        imap_user="alice@nyu.edu",
+        smtp_user="ALICE@NYU.EDU",
+    )
     _make_account(db, account_id="acct-other", owner="alice")  # must stay untouched
     db.close()
 
@@ -407,6 +415,166 @@ async def test_callback_valid_owner_writes_encrypted_tokens_to_intended_account(
     assert other.oauth_access_token is None, "tokens must only touch the intended account"
 
 
+@pytest.mark.asyncio
+async def test_callback_rejects_token_for_a_different_mailbox_identity():
+    """Reconnecting with another Google identity must not replace the token
+    while retaining the original IMAP/SMTP login names."""
+    from routes.email_helpers import make_oauth_state
+    from src.secret_storage import encrypt as _enc, decrypt as _dec
+    from core.database import EmailAccount
+
+    db, Factory = _make_db()
+    _make_account(
+        db,
+        account_id="acct-reconnect",
+        owner="alice",
+        imap_user="alice@example.edu",
+        smtp_user="alice@example.edu",
+        oauth_provider="google",
+        oauth_access_token=_enc("ya29.existing_access"),
+        oauth_refresh_token=_enc("1//existing_refresh"),
+    )
+    db.close()
+
+    token_resp = mock.MagicMock()
+    token_resp.raise_for_status = mock.MagicMock()
+    token_resp.json.return_value = {
+        "access_token": "ya29.other_access",
+        "refresh_token": "1//other_refresh",
+        "expires_in": 3600,
+    }
+    userinfo_resp = mock.MagicMock()
+    userinfo_resp.is_success = True
+    userinfo_resp.json.return_value = {
+        "email": "other@example.edu",
+        "name": "Other User",
+    }
+
+    state = make_oauth_state("acct-reconnect", "alice")
+    with mock.patch("httpx.post", return_value=token_resp), \
+         mock.patch("httpx.get", return_value=userinfo_resp), \
+         mock.patch("core.database.SessionLocal", Factory):
+        resp = await _callback_endpoint()(
+            code="4/code",
+            state=state,
+            error=None,
+            request=_FakeRequest(),
+        )
+
+    assert "email_oauth_error=identity_verification_failed" in _location(resp)
+    verify_db = Factory()
+    row = verify_db.query(EmailAccount).filter(
+        EmailAccount.id == "acct-reconnect"
+    ).first()
+    verify_db.close()
+    assert _dec(row.oauth_access_token) == "ya29.existing_access"
+    assert _dec(row.oauth_refresh_token) == "1//existing_refresh"
+
+
+@pytest.mark.asyncio
+async def test_callback_rejects_reconnect_without_a_fresh_refresh_token():
+    """A same-identity access token cannot be paired with an unproven refresh
+    token retained from a previously mixed row."""
+    from routes.email_helpers import make_oauth_state
+    from src.secret_storage import encrypt as _enc, decrypt as _dec
+    from core.database import EmailAccount
+
+    db, Factory = _make_db()
+    _make_account(
+        db,
+        account_id="acct-refresh-proof",
+        owner="alice",
+        imap_user="alice@example.edu",
+        smtp_user="alice@example.edu",
+        oauth_provider="google",
+        oauth_access_token=_enc("ya29.existing_access"),
+        oauth_refresh_token=_enc("1//refresh_for_other_identity"),
+    )
+    db.close()
+
+    token_resp = mock.MagicMock()
+    token_resp.raise_for_status = mock.MagicMock()
+    token_resp.json.return_value = {
+        "access_token": "ya29.same_identity_access",
+        "expires_in": 3600,
+    }
+
+    state = make_oauth_state("acct-refresh-proof", "alice")
+    with mock.patch("httpx.post", return_value=token_resp), \
+         mock.patch("httpx.get") as userinfo_get, \
+         mock.patch("core.database.SessionLocal", Factory):
+        resp = await _callback_endpoint()(
+            code="4/code",
+            state=state,
+            error=None,
+            request=_FakeRequest(),
+        )
+
+    assert "email_oauth_error=token_exchange_failed" in _location(resp)
+    userinfo_get.assert_not_called()
+    verify_db = Factory()
+    row = verify_db.query(EmailAccount).filter(
+        EmailAccount.id == "acct-refresh-proof"
+    ).first()
+    verify_db.close()
+    assert _dec(row.oauth_access_token) == "ya29.existing_access"
+    assert _dec(row.oauth_refresh_token) == "1//refresh_for_other_identity"
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("userinfo_result", [None, {}, {"email": None}])
+async def test_callback_requires_verified_mailbox_identity(userinfo_result):
+    """A failed or incomplete userinfo lookup must not persist fresh tokens."""
+    from routes.email_helpers import make_oauth_state
+    from core.database import EmailAccount
+
+    db, Factory = _make_db()
+    _make_account(
+        db,
+        account_id="acct-no-identity",
+        owner="alice",
+        imap_user="alice@example.edu",
+        smtp_user="alice@example.edu",
+    )
+    db.close()
+
+    token_resp = mock.MagicMock()
+    token_resp.raise_for_status = mock.MagicMock()
+    token_resp.json.return_value = {
+        "access_token": "ya29.unverified_access",
+        "refresh_token": "1//unverified_refresh",
+        "expires_in": 3600,
+    }
+    if userinfo_result is None:
+        userinfo_call = mock.Mock(side_effect=RuntimeError("userinfo unavailable"))
+    else:
+        userinfo_resp = mock.MagicMock()
+        userinfo_resp.is_success = True
+        userinfo_resp.json.return_value = userinfo_result
+        userinfo_call = mock.Mock(return_value=userinfo_resp)
+
+    state = make_oauth_state("acct-no-identity", "alice")
+    with mock.patch("httpx.post", return_value=token_resp), \
+         mock.patch("httpx.get", userinfo_call), \
+         mock.patch("core.database.SessionLocal", Factory):
+        resp = await _callback_endpoint()(
+            code="4/code",
+            state=state,
+            error=None,
+            request=_FakeRequest(),
+        )
+
+    assert "email_oauth_error=identity_verification_failed" in _location(resp)
+    verify_db = Factory()
+    row = verify_db.query(EmailAccount).filter(
+        EmailAccount.id == "acct-no-identity"
+    ).first()
+    verify_db.close()
+    assert row.oauth_provider is None
+    assert row.oauth_access_token is None
+    assert row.oauth_refresh_token is None
+
+
 # ── Token refresh scenarios ───────────────────────────────────────
 
 def test_get_valid_google_token_uses_cached_when_fresh():