diff --git a/mcp_servers/email_server.py b/mcp_servers/email_server.py index 96c985864..e2bccfbfb 100644 --- a/mcp_servers/email_server.py +++ b/mcp_servers/email_server.py @@ -58,6 +58,11 @@ def _uid_fetch_rows(data) -> list: _ACCOUNT_CACHE: dict = {} # key = normalized account selector -> config dict _MCP_OWNER_ARG = "_odysseus_owner" _CURRENT_OWNER: ContextVar[str | None] = ContextVar("email_mcp_owner", default=None) +_OWNER_ENV_KEYS = ("ODYSSEUS_MCP_EMAIL_OWNER", "ODYSSEUS_EMAIL_OWNER") +_OWNER_SCOPE_ERROR = ( + "Error: email MCP requires an authenticated owner or ODYSSEUS_MCP_EMAIL_OWNER " + "when owner-scoped email accounts are configured." +) def _clean_header_value(value) -> str: @@ -71,13 +76,29 @@ def _db_path() -> Path: return Path(APP_DB) +def _configured_owner() -> str | None: + for key in _OWNER_ENV_KEYS: + owner = os.environ.get(key, "").strip() + if owner: + return owner + return None + + def _current_owner() -> str: owner = _CURRENT_OWNER.get() - return str(owner or "").strip() + return str(owner or _configured_owner() or "").strip() + + +def _account_owner(row: dict) -> str: + return str(row.get("owner") or "").strip() + + +def _has_owner_scoped_accounts(rows: list[dict]) -> bool: + return any(_account_owner(r) for r in rows) def _account_visible_to_owner(row: dict, owner: str) -> bool: - row_owner = str(row.get("owner") or "").strip() + row_owner = _account_owner(row) if row_owner == owner: return True if row_owner: @@ -96,8 +117,7 @@ def _filter_accounts_for_owner(rows: list[dict]) -> list[dict]: if owner: return [r for r in rows if _account_visible_to_owner(r, owner)] - owners = {str(r.get("owner") or "").strip() for r in rows if str(r.get("owner") or "").strip()} - if len(owners) > 1: + if _has_owner_scoped_accounts(rows): return [] return rows @@ -106,8 +126,7 @@ def _mcp_owner_required(rows: list[dict] | None = None) -> bool: if _current_owner(): return False rows = rows if rows is not None else _read_accounts_from_db() - owners = {str(r.get("owner") or "").strip() for r in rows if str(r.get("owner") or "").strip()} - return len(owners) > 1 + return _has_owner_scoped_accounts(rows) def _load_email_writing_style() -> str: @@ -274,6 +293,8 @@ def _load_config(account: str | None = None) -> dict: } raw_rows = _read_accounts_from_db() + if _mcp_owner_required(raw_rows): + raise ValueError(_OWNER_SCOPE_ERROR) rows = _filter_accounts_for_owner(raw_rows) row = _resolve_account_from_rows(rows, account) if _current_owner() and raw_rows and not rows: @@ -1193,10 +1214,14 @@ def _send_email(to, subject, body, in_reply_to=None, references=None, cc=None, b UI. This closes the auto-send hole that let earlier models invent signatures and ship them to real recipients without confirmation.""" if _read_agent_email_confirm_setting(): + # Even confirmation-first sends must resolve the selected account now. + # Otherwise a caller could stage a pending draft against another + # owner's account selector before browser approval handles it. + cfg = _load_config(account) return _stash_agent_draft( to=to, subject=subject, body=body, in_reply_to=in_reply_to, references=references, - cc=cc, bcc=bcc, account=account, + cc=cc, bcc=bcc, account=cfg.get("account_id") or account, ) send_account, cfg = _resolve_send_config(account) msg = EmailMessage() @@ -2142,10 +2167,7 @@ async def call_tool(name: str, arguments: dict) -> list[TextContent]: try: all_db_accounts = _read_accounts_from_db() if _mcp_owner_required(all_db_accounts): - return [TextContent( - type="text", - text="Error: email MCP requires an authenticated owner when multiple email account owners are configured.", - )] + return [TextContent(type="text", text=_OWNER_SCOPE_ERROR)] if name == "list_email_accounts": rows = _filter_accounts_for_owner(all_db_accounts) diff --git a/tests/test_mcp_email_decode_header_spaces.py b/tests/test_mcp_email_decode_header_spaces.py index f588a6bd3..6370801b0 100644 --- a/tests/test_mcp_email_decode_header_spaces.py +++ b/tests/test_mcp_email_decode_header_spaces.py @@ -16,7 +16,20 @@ pytest.importorskip("mcp") import mcp_servers.email_server as es -def _init_accounts_db(path): +@pytest.fixture(autouse=True) +def _clear_mcp_email_owner_env(monkeypatch): + for key in es._OWNER_ENV_KEYS: + monkeypatch.delenv(key, raising=False) + es._ACCOUNT_CACHE.clear() + yield + es._ACCOUNT_CACHE.clear() + + +def _init_accounts_db(path, rows=None): + rows = rows or [ + ("acct-alice", "alice", "Alice Mail", 1, "alice@example.com", "alice@example.com", "alice@example.com", "2026-01-01"), + ("acct-bob", "bob", "Bob Mail", 1, "bob@example.com", "bob@example.com", "bob@example.com", "2026-01-02"), + ] conn = sqlite3.connect(path) conn.execute( """ @@ -50,10 +63,7 @@ def _init_accounts_db(path): VALUES (?, ?, ?, ?, 1, 'imap.example.com', 993, ?, '', 1, 'smtp.example.com', 465, 'ssl', ?, '', ?, ?) """, - [ - ("acct-alice", "alice", "Alice Mail", 1, "alice@example.com", "alice@example.com", "alice@example.com", "2026-01-01"), - ("acct-bob", "bob", "Bob Mail", 1, "bob@example.com", "bob@example.com", "bob@example.com", "2026-01-02"), - ], + rows, ) conn.commit() conn.close() @@ -106,6 +116,37 @@ async def test_mcp_email_requires_owner_when_multiple_account_owners_exist(tmp_p assert "requires an authenticated owner" in out[0].text +@pytest.mark.asyncio +async def test_mcp_email_requires_owner_when_any_account_owner_exists(tmp_path, monkeypatch): + db_path = tmp_path / "app.db" + _init_accounts_db( + db_path, + rows=[ + ("acct-alice", "alice", "Alice Mail", 1, "alice@example.com", "alice@example.com", "alice@example.com", "2026-01-01"), + ], + ) + monkeypatch.setattr(es, "APP_DB", str(db_path)) + + out = await es.call_tool("list_email_accounts", {}) + + assert "requires an authenticated owner" in out[0].text + assert "Alice Mail" not in out[0].text + + +@pytest.mark.asyncio +async def test_mcp_email_configured_owner_filters_accounts(tmp_path, monkeypatch): + db_path = tmp_path / "app.db" + _init_accounts_db(db_path) + monkeypatch.setattr(es, "APP_DB", str(db_path)) + monkeypatch.setenv("ODYSSEUS_MCP_EMAIL_OWNER", "alice") + + out = await es.call_tool("list_email_accounts", {}) + text = out[0].text + + assert "Alice Mail" in text + assert "Bob Mail" not in text + + def test_mcp_email_scoped_owner_without_visible_account_skips_legacy_fallback(tmp_path, monkeypatch): db_path = tmp_path / "app.db" settings_path = tmp_path / "settings.json" @@ -137,11 +178,81 @@ def test_mcp_email_scoped_owner_without_visible_account_skips_legacy_fallback(tm es._ACCOUNT_CACHE.clear() +@pytest.mark.asyncio +async def test_mcp_email_owner_cannot_use_other_owner_account_for_list_read_send_or_draft(tmp_path, monkeypatch): + import src.constants as constants + + db_path = tmp_path / "app.db" + scheduled_path = tmp_path / "scheduled_emails.db" + _init_accounts_db(db_path) + monkeypatch.setattr(es, "APP_DB", str(db_path)) + monkeypatch.setattr(constants, "SCHEDULED_EMAILS_DB", str(scheduled_path)) + monkeypatch.setattr(es, "_read_agent_email_confirm_setting", lambda: True) + + calls = [ + ("list_emails", {"account": "Bob Mail"}), + ("read_email", {"uid": "1", "account": "Bob Mail"}), + ("send_email", { + "to": "recipient@example.com", + "subject": "Blocked", + "body": "Do not stage.", + "account": "Bob Mail", + }), + ("draft_email", { + "to": "recipient@example.com", + "subject": "Blocked", + "body": "Do not draft.", + "account": "Bob Mail", + }), + ] + + for tool_name, args in calls: + out = await es.call_tool(tool_name, {**args, "_odysseus_owner": "alice"}) + assert "Email account not found for selector" in out[0].text, tool_name + assert "Bob Mail" not in out[0].text or "Available accounts" in out[0].text + + assert not scheduled_path.exists() + + +@pytest.mark.asyncio +async def test_mcp_send_email_stages_with_visible_owner_account_id(tmp_path, monkeypatch): + import src.constants as constants + + app_db_path = tmp_path / "app.db" + scheduled_path = tmp_path / "scheduled_emails.db" + _init_accounts_db(app_db_path) + monkeypatch.setattr(es, "APP_DB", str(app_db_path)) + monkeypatch.setattr(constants, "SCHEDULED_EMAILS_DB", str(scheduled_path)) + monkeypatch.setattr(es, "_read_agent_email_confirm_setting", lambda: True) + + out = await es.call_tool( + "send_email", + { + "to": "recipient@example.com", + "subject": "Review", + "body": "Please review.", + "account": "Alice Mail", + "_odysseus_owner": "alice", + }, + ) + + assert "Draft staged for approval" in out[0].text + conn = sqlite3.connect(scheduled_path) + try: + row = conn.execute( + "SELECT owner, status, account_id FROM scheduled_emails" + ).fetchone() + finally: + conn.close() + assert row == ("alice", "agent_draft", "acct-alice") + + @pytest.mark.asyncio async def test_mcp_send_email_stages_owner_scoped_pending_draft(tmp_path, monkeypatch): import src.constants as constants db_path = tmp_path / "scheduled_emails.db" + monkeypatch.setattr(es, "APP_DB", str(tmp_path / "missing-app.db")) monkeypatch.setattr(constants, "SCHEDULED_EMAILS_DB", str(db_path)) monkeypatch.setattr(es, "_read_agent_email_confirm_setting", lambda: True)