diff --git a/src/openjarvis/connectors/gmail.py b/src/openjarvis/connectors/gmail.py new file mode 100644 index 00000000..d486cfb1 --- /dev/null +++ b/src/openjarvis/connectors/gmail.py @@ -0,0 +1,393 @@ +"""Gmail connector — bulk email sync via the Gmail REST API. + +Uses OAuth 2.0 tokens stored locally (see :mod:`openjarvis.connectors.oauth`). +All network calls are isolated in module-level functions (``_gmail_api_*``) +to make them trivially mockable in tests. +""" + +from __future__ import annotations + +import base64 +import email.utils +from datetime import datetime +from typing import Any, Dict, Iterator, List, Optional + +import httpx + +from openjarvis.connectors._stubs import BaseConnector, Document, SyncStatus +from openjarvis.connectors.oauth import ( + build_google_auth_url, + delete_tokens, + load_tokens, + save_tokens, +) +from openjarvis.core.config import DEFAULT_CONFIG_DIR +from openjarvis.core.registry import ConnectorRegistry +from openjarvis.tools._stubs import ToolSpec + +# --------------------------------------------------------------------------- +# Constants +# --------------------------------------------------------------------------- + +_GMAIL_API_BASE = "https://gmail.googleapis.com/gmail/v1/users/me" +_GMAIL_SCOPE = "https://www.googleapis.com/auth/gmail.readonly" +_DEFAULT_CREDENTIALS_PATH = str( + DEFAULT_CONFIG_DIR / "connectors" / "gmail.json" +) + +# --------------------------------------------------------------------------- +# Module-level API functions (easy to patch in tests) +# --------------------------------------------------------------------------- + + +def _gmail_api_list_messages( + token: str, + *, + page_token: Optional[str] = None, + query: str = "", +) -> Dict[str, Any]: + """Call the Gmail ``messages.list`` endpoint. + + Parameters + ---------- + token: + OAuth access token. + page_token: + Pagination token from a previous response's ``nextPageToken``. + query: + Gmail search query string (e.g. ``"is:unread"``). + + Returns + ------- + dict + Raw API response containing ``messages`` list and optional + ``nextPageToken``. + """ + params: Dict[str, str] = {} + if page_token: + params["pageToken"] = page_token + if query: + params["q"] = query + + resp = httpx.get( + f"{_GMAIL_API_BASE}/messages", + headers={"Authorization": f"Bearer {token}"}, + params=params, + timeout=30.0, + ) + resp.raise_for_status() + return resp.json() + + +def _gmail_api_get_message(token: str, msg_id: str) -> Dict[str, Any]: + """Fetch a single Gmail message by ID (``full`` format). + + Parameters + ---------- + token: + OAuth access token. + msg_id: + Gmail message ID string. + + Returns + ------- + dict + Raw API response for the message resource. + """ + resp = httpx.get( + f"{_GMAIL_API_BASE}/messages/{msg_id}", + headers={"Authorization": f"Bearer {token}"}, + params={"format": "full"}, + timeout=30.0, + ) + resp.raise_for_status() + return resp.json() + + +# --------------------------------------------------------------------------- +# Helper functions +# --------------------------------------------------------------------------- + + +def _extract_header(headers: List[Dict[str, str]], name: str) -> str: + """Return the value of the first header matching *name* (case-insensitive).""" + name_lower = name.lower() + for h in headers: + if h.get("name", "").lower() == name_lower: + return h.get("value", "") + return "" + + +def _decode_body(payload: Dict[str, Any]) -> str: + """Decode the message body from a Gmail payload dict. + + Handles both simple payloads (``body.data``) and multipart messages + by recursively searching for a ``text/plain`` part. + """ + mime_type: str = payload.get("mimeType", "") + + if mime_type.startswith("multipart/"): + # Search parts for text/plain first, then any text/* fallback + parts: List[Dict[str, Any]] = payload.get("parts", []) + for part in parts: + if part.get("mimeType", "").startswith("text/plain"): + return _decode_body(part) + # Fallback: recurse into first part + if parts: + return _decode_body(parts[0]) + return "" + + body_data: str = payload.get("body", {}).get("data", "") + if not body_data: + return "" + + # Gmail uses URL-safe base64 without padding + padded = body_data + "=" * (-len(body_data) % 4) + try: + return base64.urlsafe_b64decode(padded).decode("utf-8", errors="replace") + except Exception: # noqa: BLE001 + return "" + + +def _parse_date(date_str: str) -> datetime: + """Parse an RFC 2822 email date string into a :class:`~datetime.datetime`. + + Falls back to :func:`datetime.now` if the string is unparseable. + """ + if not date_str: + return datetime.now() + try: + return email.utils.parsedate_to_datetime(date_str) + except Exception: # noqa: BLE001 + return datetime.now() + + +# --------------------------------------------------------------------------- +# GmailConnector +# --------------------------------------------------------------------------- + + +@ConnectorRegistry.register("gmail") +class GmailConnector(BaseConnector): + """Connector that syncs emails from Gmail via the REST API. + + Authentication is handled through Google OAuth 2.0. Tokens are stored + locally in a JSON credentials file. + + Parameters + ---------- + credentials_path: + Path to the JSON file where OAuth tokens are stored. Defaults to + ``~/.openjarvis/connectors/gmail.json``. + """ + + connector_id = "gmail" + display_name = "Gmail" + auth_type = "oauth" + + def __init__(self, credentials_path: str = "") -> None: + self._credentials_path = credentials_path or _DEFAULT_CREDENTIALS_PATH + self._items_synced: int = 0 + self._items_total: int = 0 + self._last_sync: Optional[datetime] = None + self._last_cursor: Optional[str] = None + + # ------------------------------------------------------------------ + # BaseConnector interface + # ------------------------------------------------------------------ + + def is_connected(self) -> bool: + """Return ``True`` if a credentials file with a valid token exists.""" + tokens = load_tokens(self._credentials_path) + if tokens is None: + return False + # Accept any non-empty dict that contains at least one key + # (simplified: real impl would also check expiry / refresh token) + return bool(tokens) + + def disconnect(self) -> None: + """Delete the stored credentials file.""" + delete_tokens(self._credentials_path) + + def auth_url(self) -> str: + """Return a Google OAuth consent URL requesting ``gmail.readonly`` scope.""" + return build_google_auth_url( + client_id="", # placeholder — real client_id from config + scopes=[_GMAIL_SCOPE], + ) + + def handle_callback(self, code: str) -> None: + """Handle the OAuth callback by persisting the authorization code. + + In a full implementation this would exchange the code for tokens. + For now the code is saved directly as the token value. + """ + save_tokens(self._credentials_path, {"token": code}) + + def sync( + self, + *, + since: Optional[datetime] = None, # noqa: ARG002 — reserved for future use + cursor: Optional[str] = None, + ) -> Iterator[Document]: + """Yield :class:`Document` objects for Gmail messages. + + Paginates through the messages.list API and fetches each message's + full payload to extract headers and body. + + Parameters + ---------- + since: + Not yet used (Gmail API filtering is done server-side via query). + cursor: + ``nextPageToken`` from a previous sync to resume pagination. + """ + tokens = load_tokens(self._credentials_path) + if not tokens: + return + + token: str = tokens.get("token", tokens.get("access_token", "")) + if not token: + return + + page_token: Optional[str] = cursor + synced = 0 + + while True: + list_resp = _gmail_api_list_messages(token, page_token=page_token) + messages: List[Dict[str, Any]] = list_resp.get("messages", []) + + for msg_stub in messages: + msg_id: str = msg_stub.get("id", "") + if not msg_id: + continue + + msg = _gmail_api_get_message(token, msg_id) + payload: Dict[str, Any] = msg.get("payload", {}) + headers: List[Dict[str, str]] = payload.get("headers", []) + + from_header = _extract_header(headers, "From") + subject = _extract_header(headers, "Subject") + date_str = _extract_header(headers, "Date") + to_header = _extract_header(headers, "To") + + body = _decode_body(payload) + timestamp = _parse_date(date_str) + + participants: List[str] = [] + if from_header: + participants.append(from_header) + if to_header: + participants.append(to_header) + + thread_id: Optional[str] = msg.get("threadId") + + doc = Document( + doc_id=f"gmail:{msg_id}", + source="gmail", + doc_type="email", + content=body, + title=subject, + author=from_header, + participants=participants, + timestamp=timestamp, + thread_id=thread_id, + metadata={ + "message_id": msg_id, + "labels": msg.get("labelIds", []), + }, + ) + synced += 1 + yield doc + + next_page: Optional[str] = list_resp.get("nextPageToken") + if not next_page: + self._last_cursor = None + break + page_token = next_page + self._last_cursor = next_page + + self._items_synced = synced + self._last_sync = datetime.now() + + def sync_status(self) -> SyncStatus: + """Return sync progress from the most recent :meth:`sync` call.""" + return SyncStatus( + state="idle", + items_synced=self._items_synced, + last_sync=self._last_sync, + cursor=self._last_cursor, + ) + + # ------------------------------------------------------------------ + # MCP tools + # ------------------------------------------------------------------ + + def mcp_tools(self) -> List[ToolSpec]: + """Expose three MCP tool specs for real-time Gmail queries.""" + return [ + ToolSpec( + name="gmail_search_emails", + description=( + "Search Gmail messages using a query string. " + "Supports the same syntax as the Gmail search box " + "(e.g. 'from:alice subject:report is:unread')." + ), + parameters={ + "type": "object", + "properties": { + "query": { + "type": "string", + "description": "Gmail search query", + }, + "max_results": { + "type": "integer", + "description": "Maximum number of emails to return", + "default": 20, + }, + }, + "required": ["query"], + }, + category="communication", + ), + ToolSpec( + name="gmail_get_thread", + description=( + "Retrieve all messages in a Gmail thread by thread ID." + ), + parameters={ + "type": "object", + "properties": { + "thread_id": { + "type": "string", + "description": "Gmail thread ID", + }, + }, + "required": ["thread_id"], + }, + category="communication", + ), + ToolSpec( + name="gmail_list_unread", + description=( + "List unread Gmail messages, optionally filtered by label." + ), + parameters={ + "type": "object", + "properties": { + "label": { + "type": "string", + "description": "Gmail label to filter by (e.g. 'INBOX')", + "default": "INBOX", + }, + "max_results": { + "type": "integer", + "description": "Maximum number of messages to return", + "default": 20, + }, + }, + "required": [], + }, + category="communication", + ), + ] diff --git a/src/openjarvis/connectors/oauth.py b/src/openjarvis/connectors/oauth.py new file mode 100644 index 00000000..a32b7262 --- /dev/null +++ b/src/openjarvis/connectors/oauth.py @@ -0,0 +1,97 @@ +"""Shared OAuth 2.0 helpers for Google connectors. + +Provides URL builder, token persistence, and token cleanup utilities +that are reused by gmail, drive, calendar, and contacts connectors. +""" + +from __future__ import annotations + +import json +import os +from pathlib import Path +from typing import Any, Dict, List, Optional +from urllib.parse import urlencode + +# --------------------------------------------------------------------------- +# Google OAuth endpoints / defaults +# --------------------------------------------------------------------------- + +_GOOGLE_AUTH_ENDPOINT = "https://accounts.google.com/o/oauth2/v2/auth" +_DEFAULT_REDIRECT_URI = "http://localhost:8789/callback" +_DEFAULT_SCOPES: List[str] = ["openid", "email", "profile"] + + +# --------------------------------------------------------------------------- +# Public API +# --------------------------------------------------------------------------- + + +def build_google_auth_url( + client_id: str, + redirect_uri: str = _DEFAULT_REDIRECT_URI, + scopes: Optional[List[str]] = None, +) -> str: + """Build a Google OAuth2 consent URL. + + Parameters + ---------- + client_id: + The OAuth 2.0 client ID from the Google Cloud Console. + redirect_uri: + Where Google should redirect after consent. Defaults to the local + callback server at ``http://localhost:8789/callback``. + scopes: + List of OAuth scopes to request. Defaults to + ``["openid", "email", "profile"]``. + + Returns + ------- + str + Full consent URL including query string. + """ + if scopes is None: + scopes = _DEFAULT_SCOPES + + params = { + "client_id": client_id, + "redirect_uri": redirect_uri, + "response_type": "code", + "scope": " ".join(scopes), + "access_type": "offline", + "prompt": "consent", + } + return f"{_GOOGLE_AUTH_ENDPOINT}?{urlencode(params)}" + + +def load_tokens(path: str) -> Optional[Dict[str, Any]]: + """Load OAuth tokens from a JSON file. + + Returns ``None`` if the file is missing, unreadable, or contains + invalid JSON. + """ + p = Path(path) + if not p.exists(): + return None + try: + raw = p.read_text(encoding="utf-8") + return json.loads(raw) + except (OSError, json.JSONDecodeError): + return None + + +def save_tokens(path: str, tokens: Dict[str, Any]) -> None: + """Persist *tokens* to *path* as JSON with owner-only (0o600) permissions. + + Creates parent directories as needed. + """ + p = Path(path) + p.parent.mkdir(parents=True, exist_ok=True) + p.write_text(json.dumps(tokens, indent=2), encoding="utf-8") + os.chmod(path, 0o600) + + +def delete_tokens(path: str) -> None: + """Delete the credentials file at *path* if it exists.""" + p = Path(path) + if p.exists(): + p.unlink() diff --git a/tests/connectors/test_gmail.py b/tests/connectors/test_gmail.py new file mode 100644 index 00000000..e7272774 --- /dev/null +++ b/tests/connectors/test_gmail.py @@ -0,0 +1,217 @@ +"""Tests for GmailConnector — OAuth-authenticated Gmail sync connector. + +All Gmail API calls are mocked; no network access is required. +""" + +from __future__ import annotations + +import json +from pathlib import Path +from typing import List +from unittest.mock import patch + +import pytest + +from openjarvis.connectors._stubs import Document +from openjarvis.core.registry import ConnectorRegistry + +# --------------------------------------------------------------------------- +# Helpers — fake API payloads +# --------------------------------------------------------------------------- + +# base64url("Hello world") == "SGVsbG8gd29ybGQ=" +# base64url("Budget reply") == "QnVkZ2V0IHJlcGx5" + +_MSG1 = { + "id": "msg1", + "threadId": "thread1", + "labelIds": ["INBOX"], + "payload": { + "mimeType": "text/plain", + "headers": [ + {"name": "From", "value": "alice@example.com"}, + {"name": "To", "value": "me@example.com"}, + {"name": "Subject", "value": "Q3 Planning"}, + {"name": "Date", "value": "Mon, 01 Jan 2024 10:00:00 +0000"}, + ], + "body": {"data": "SGVsbG8gd29ybGQ="}, + }, +} + +_MSG2 = { + "id": "msg2", + "threadId": "thread2", + "labelIds": ["INBOX"], + "payload": { + "mimeType": "text/plain", + "headers": [ + {"name": "From", "value": "bob@example.com"}, + {"name": "To", "value": "me@example.com"}, + {"name": "Subject", "value": "Re: Budget"}, + {"name": "Date", "value": "Tue, 02 Jan 2024 12:00:00 +0000"}, + ], + "body": {"data": "QnVkZ2V0IHJlcGx5"}, + }, +} + +_LIST_RESPONSE = { + "messages": [{"id": "msg1"}, {"id": "msg2"}], + # No nextPageToken → single page +} + + +def _make_credentials(tmp_path: Path) -> Path: + """Write a minimal fake credentials file and return its path.""" + creds = tmp_path / "gmail.json" + creds.write_text(json.dumps({"token": "fake-access-token"}), encoding="utf-8") + return creds + + +# --------------------------------------------------------------------------- +# Fixtures +# --------------------------------------------------------------------------- + + +@pytest.fixture() +def connector(tmp_path: Path): + """GmailConnector pointing at a tmp credentials path (no file yet).""" + from openjarvis.connectors.gmail import GmailConnector # noqa: PLC0415 + + creds_path = str(tmp_path / "gmail.json") + return GmailConnector(credentials_path=creds_path) + + +# --------------------------------------------------------------------------- +# Test 1 — not connected without a credentials file +# --------------------------------------------------------------------------- + + +def test_not_connected_without_credentials(connector) -> None: + """is_connected() returns False when no credentials file exists.""" + assert connector.is_connected() is False + + +# --------------------------------------------------------------------------- +# Test 2 — auth_type is "oauth" +# --------------------------------------------------------------------------- + + +def test_auth_type_is_oauth(connector) -> None: + """GmailConnector.auth_type must be 'oauth'.""" + assert connector.auth_type == "oauth" + + +# --------------------------------------------------------------------------- +# Test 3 — auth_url returns a valid Google consent URL +# --------------------------------------------------------------------------- + + +def test_auth_url_returns_string(connector) -> None: + """auth_url() returns a URL pointing to Google's OAuth endpoint.""" + url = connector.auth_url() + assert isinstance(url, str) + assert url.startswith("https://accounts.google.com/o/oauth2/v2/auth") + assert "gmail.readonly" in url + + +# --------------------------------------------------------------------------- +# Test 4 — sync yields documents with correct fields (mocked API) +# --------------------------------------------------------------------------- + + +@patch("openjarvis.connectors.gmail._gmail_api_list_messages") +@patch("openjarvis.connectors.gmail._gmail_api_get_message") +def test_sync_yields_documents( + mock_get, + mock_list, + connector, + tmp_path: Path, +) -> None: + """sync() yields one Document per message with correct metadata.""" + # Set up fake credentials so is_connected() returns True + creds_path = Path(connector._credentials_path) + creds_path.write_text( + json.dumps({"token": "fake-access-token"}), encoding="utf-8" + ) + + # Configure mocks + mock_list.return_value = _LIST_RESPONSE + mock_get.side_effect = lambda token, msg_id: ( + _MSG1 if msg_id == "msg1" else _MSG2 + ) + + docs: List[Document] = list(connector.sync()) + + assert len(docs) == 2 + + # --- Message 1 --- + doc1 = next(d for d in docs if d.doc_id == "gmail:msg1") + assert doc1.source == "gmail" + assert doc1.doc_type == "email" + assert doc1.title == "Q3 Planning" + assert doc1.author == "alice@example.com" + assert doc1.content == "Hello world" + assert doc1.thread_id == "thread1" + assert "alice@example.com" in doc1.participants + + # --- Message 2 --- + doc2 = next(d for d in docs if d.doc_id == "gmail:msg2") + assert doc2.title == "Re: Budget" + assert doc2.author == "bob@example.com" + assert doc2.content == "Budget reply" + assert doc2.thread_id == "thread2" + + # Verify the API was called correctly + mock_list.assert_called_once() + assert mock_get.call_count == 2 + + +# --------------------------------------------------------------------------- +# Test 5 — disconnect removes the credentials file +# --------------------------------------------------------------------------- + + +def test_disconnect(connector, tmp_path: Path) -> None: + """disconnect() deletes the credentials file.""" + creds_path = Path(connector._credentials_path) + creds_path.write_text( + json.dumps({"token": "fake-access-token"}), encoding="utf-8" + ) + assert connector.is_connected() is True + + connector.disconnect() + + assert not creds_path.exists() + assert connector.is_connected() is False + + +# --------------------------------------------------------------------------- +# Test 6 — mcp_tools returns the three expected tool specs +# --------------------------------------------------------------------------- + + +def test_mcp_tools(connector) -> None: + """mcp_tools() returns exactly 3 tools with the required names.""" + tools = connector.mcp_tools() + names = {t.name for t in tools} + assert len(tools) == 3 + assert "gmail_search_emails" in names + assert "gmail_get_thread" in names + assert "gmail_list_unread" in names + + +# --------------------------------------------------------------------------- +# Test 7 — ConnectorRegistry contains "gmail" after import +# --------------------------------------------------------------------------- + + +def test_registry() -> None: + """GmailConnector can be registered and retrieved via ConnectorRegistry.""" + from openjarvis.connectors.gmail import GmailConnector # noqa: PLC0415 + + # The registry is cleared before each test by the autouse conftest fixture, + # so we imperatively re-register here (same pattern as test_obsidian.py). + ConnectorRegistry.register_value("gmail", GmailConnector) + assert ConnectorRegistry.contains("gmail") + cls = ConnectorRegistry.get("gmail") + assert cls.connector_id == "gmail"