diff --git a/.env.example b/.env.example index a6e63d9..f874b83 100644 --- a/.env.example +++ b/.env.example @@ -30,6 +30,17 @@ IMAP_PORT=993 IMAP_USERNAME= IMAP_PASSWORD= +# Option B2: Microsoft 365 / Outlook [for a mailbox on Exchange Online] +# Set provider: microsoft in harvey.yaml. Exchange Online no longer +# accepts passwords over IMAP, so this signs in with OAuth instead: +# 1. entra.microsoft.com → App registrations → New registration +# ("this organizational directory only"; redirect URI platform +# "Public client/native", value http://localhost:8766) +# 2. Paste the Application (client) ID and Directory (tenant) ID below +# 3. Run: harvey microsoft auth (then: harvey microsoft test) +M365_CLIENT_ID= +M365_TENANT_ID= + # Option C: Instantly — cold email platform [legacy] # Where to get it: https://app.instantly.ai → Settings → Integrations → API Keys # API access requires Instantly's Growth plan or higher. diff --git a/CHANGELOG.md b/CHANGELOG.md index 73137aa..db180c5 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,15 @@ minor versions can still change behaviour. ### Added +- **Microsoft 365 mail provider.** Set `channels.email.provider: microsoft` to + send and read through a Microsoft 365 / Outlook mailbox via Microsoft Graph, + for domains whose mail lives in Exchange Online (which no longer accepts + passwords over IMAP). Signs in once with OAuth + PKCE (`harvey microsoft + auth`; no client secret, no stored password), drafts then sends so the + Message-ID is known, threads replies with `createReply`, reads the last 7 + days of the inbox without marking anything read, refuses to act as any + mailbox other than `persona.email`, and holds sending when the inbox can't + be read. See [docs/microsoft365.md](docs/microsoft365.md). - **Optional per-recipient compliance check (Precleared).** Set `compliance.precleared.enabled: true` with `sender_country: GB` or `US` to have each new recipient checked before the first cold email against UK diff --git a/README.md b/README.md index 581cd19..3faf678 100644 --- a/README.md +++ b/README.md @@ -107,7 +107,7 @@ Profiling and outbox draining ride along on **every** cycle regardless, because - **Python 3.11+** — check with `python3 --version`. macOS ships 3.9; if that's what you have, `brew install python@3.13` first. - An active **Claude Pro or Max** subscription, with the CLI logged in (`claude login`) — or, opt-in, a **ChatGPT plan with the Codex CLI** ([see below](#using-a-chatgpt-plan-instead-codex)) -- A mailbox to send from — Gmail/Workspace recommended, on a *dedicated secondary domain* +- A mailbox to send from — Gmail/Workspace recommended, on a *dedicated secondary domain*. Microsoft 365 works too ([setup](docs/microsoft365.md)) ### Easiest: let Claude set it up @@ -392,7 +392,7 @@ persona: { name, company, role, email, tone } product: { name, description, pricing, key_benefits, objection_responses, offer } icp: { industries, titles, company_size, geography, hiring_signals, geo_coordinates } channels: - email: { provider: gmail | smtp | instantly, max_daily_sends, require_approval, + email: { provider: gmail | microsoft | smtp | instantly, max_daily_sends, require_approval, send_to_risky, max_bounce_rate } usage: { max_daily_claude_percent, heartbeat_interval_minutes, quiet_hours } brain: { provider: claude | codex, model, reasoning_effort } # optional; claude by default @@ -417,6 +417,7 @@ harvey outbox # review queued email harvey export # deliverable prospects → CSV harvey sending pause|resume # kill switch harvey gmail auth # one-time Gmail OAuth +harvey microsoft auth|test # one-time Microsoft 365 sign-in (docs/microsoft365.md) ``` @@ -551,7 +552,7 @@ tests/ # 242 tests ## Roadmap -**Done:** heartbeat loop · observation data model with governed vocabulary · user-confirmed signal catalog · cohort builder · discovery provider menu with cost estimates · free HTTP profiler with agency detection · pattern-first email finding with honest status · native mail providers (Gmail API, SMTP+IMAP) · outbox approval ladder · deterministic pre-send gate · bounce detection and kill switch · reply handling with intent classification · subscription quota tracking · website trainer · web dashboard · CSV export +**Done:** heartbeat loop · observation data model with governed vocabulary · user-confirmed signal catalog · cohort builder · discovery provider menu with cost estimates · free HTTP profiler with agency detection · pattern-first email finding with honest status · native mail providers (Gmail API, Microsoft 365 via Graph, SMTP+IMAP) · outbox approval ladder · deterministic pre-send gate · bounce detection and kill switch · reply handling with intent classification · subscription quota tracking · website trainer · web dashboard · CSV export **Next:** people enrichment from public registries · consent-gated voice callbacks · calendar integration for auto-booking · scheduled re-observation so signal *changes* trigger outreach · multi-product support diff --git a/docs/microsoft365.md b/docs/microsoft365.md new file mode 100644 index 0000000..5639f01 --- /dev/null +++ b/docs/microsoft365.md @@ -0,0 +1,184 @@ +# Sending from a Microsoft 365 mailbox + +Harvey can send and read mail through a Microsoft 365 / Outlook mailbox +(Exchange Online) using Microsoft Graph. Use it when your sending domain's +mail lives in Microsoft 365. + +## Why not SMTP + IMAP? + +Exchange Online no longer accepts a password for IMAP, and SMTP AUTH is off +by default in most tenants. The `smtp` provider can sometimes still send, but +it can't read the inbox. Harvey has to read replies to stop sequences, honor +opt-outs and catch bounces. If it can send but can't read, it keeps emailing +people who already answered. + +The `microsoft` provider signs in with OAuth (authorization code + PKCE) +instead. Harvey never stores a password or a client secret. + +## Setup + +### 1. Register an app in Entra ID + +At [entra.microsoft.com](https://entra.microsoft.com), go to +**Applications → App registrations → New registration**: + +| Field | Value | +|---|---| +| Name | `Harvey` | +| Supported account types | *Accounts in this organizational directory only* | +| Redirect URI, platform | **Public client/native (mobile & desktop)** | +| Redirect URI, value | `http://localhost:8766` | + +From the app's **Overview**, copy the **Application (client) ID** and the +**Directory (tenant) ID**. + +Don't create a client secret. Harvey is a public client. + +### 2. Configure Harvey + +`.env`: + +``` +M365_CLIENT_ID= +M365_TENANT_ID= +``` + +`harvey.yaml` (or `harvey.local.yaml`): + +```yaml +persona: + email: "you@yourdomain.com" # must be the mailbox you sign in with +channels: + email: + provider: "microsoft" +``` + +### 3. Sign in once + +```bash +harvey microsoft auth # opens the browser; sign in as persona.email +harvey microsoft test # → ✓ Connected as you@yourdomain.com +``` + +Harvey asks for four **delegated** permissions: + +| Permission | Why it's needed | +|---|---| +| `Mail.ReadWrite` | read replies, and create drafts (reply threading) | +| `Mail.Send` | send them | +| `User.Read` | check which mailbox the token belongs to | +| `offline_access` | keep working without signing in again | + +The refresh token is stored in `data/m365_token.json` (gitignored). On macOS +and Linux the file is mode 600 (owner only). Windows has no such mode bits: +the file is protected only by the access permissions (ACLs) of the folder +Harvey lives in, so keep Harvey in a folder only you can read. +Microsoft rotates it on every use, and Harvey saves the new one each time. + +### If you see "Need admin approval" + +Your tenant doesn't let users consent to apps themselves. An admin has to +approve the app once: + +1. In Entra, open **App registrations → Harvey → API permissions**. Add the + four Microsoft Graph delegated permissions above, then click + **Grant admin consent**. +2. Run `harvey microsoft auth` again and sign in as the **sending mailbox**, + not as the admin. Harvey always acts as whoever signs in. + +To make sure nobody else in the tenant can use the app, go to +**Enterprise applications → Harvey**: + +1. **Users and groups**: assign only the sending mailbox. +2. **Properties**: set **Assignment required?** to **Yes**. Do this after + step 1, or the sending mailbox is locked out too. + +## How it behaves + +- **Sending.** Harvey creates a draft and then sends it, so the email's + RFC `Message-ID` is known. Bounces and replies are matched on it. Bodies + are plain text. +- **Replies are threaded.** Harvey answers with `createReply` on the + prospect's own message, so `In-Reply-To` and `References` are correct. If + that message was deleted, the reply goes out as a new email. +- **Your inbox is left alone.** Each cycle, Harvey reads the last 7 days of + the inbox, newest first, all of it (page by page), and skips your own mail. It never marks anything + read, moves it or deletes it. It keeps its own record of what it has + handled. +- **It only acts as the right mailbox.** If the signed-in mailbox isn't + `persona.email`, every read and send is refused with a message saying so. + This catches the common mistake of signing in as the admin who approved + the app. +- **Failures hold sending.** If the inbox can't be read (expired login, + Graph outage, throttling), the cycle sends nothing. A reply nobody read + can't have stopped its follow-ups. Graph throttling (429/503 with a short + `Retry-After`) is retried once. A read that fails part-way through counts + as a failed read: Harvey never acts on part of the inbox. +- **Unsent drafts are cleaned up.** If a send fails, its draft is deleted + so it doesn't sit in your Drafts folder. +- **The sender name is the mailbox's.** Mail goes out under the mailbox's + display name in Microsoft 365, not `persona.name`. + +## Troubleshooting + +| Symptom | Fix | +|---|---| +| `Timed out waiting for the browser round-trip` | The browser never came back to `localhost:8766`. Check that nothing else uses the port, and that the redirect URI is `http://localhost:8766` under *Mobile and desktop applications* (not *Web*). | +| `AADSTS50011` (redirect URI mismatch) | Same as above: the URI and its platform must match exactly. | +| `Need admin approval` | See [If you see "Need admin approval"](#if-you-see-need-admin-approval). | +| `signed in as X, but persona.email is Y` | Run `harvey microsoft auth` again and pick the right account. Use "Use another account" if the browser is already signed in as someone else. | +| `token refresh rejected` | The login expired (90 days unused) or was revoked. Run `harvey microsoft auth` again. | + +To revoke Harvey's access, go to +[myaccount.microsoft.com](https://myaccount.microsoft.com) → **Apps**, or +delete the app registration. Then delete `data/m365_token.json`. + +## Deliverability notes + +- **Enable DKIM** for your domain in Microsoft Defender (Email + authentication → DKIM). Without it, mail from Exchange Online isn't + signed with your domain. +- **Mind the tenant.** Exchange Online watches outbound spam for the whole + tenant. If you share it with an organization, its sending reputation + depends on yours: keep `max_daily_sends` low and `require_approval: true`. + +## For developers + +| File | What it is | +|---|---| +| `harvey/integrations/microsoft.py` | `MicrosoftProvider` (a `MailProvider`) and `run_auth_flow` | +| `harvey/integrations/mail_provider.py` | Registers `microsoft` in `NATIVE_PROVIDERS` and `get_mail_provider` | +| `harvey/cli.py` | `harvey microsoft auth\|test` | +| `tests/test_microsoft.py` | Graph and login faked with `httpx.MockTransport`; no network | + +`get_replies` **raises** `GraphError` when the inbox can't be read, instead +of returning `[]`. The handler turns that into a failed read, and the +heartbeat holds sending. It returns the whole 7-day window or nothing: + +- **Every page.** `limit` is only the page size (`$top`). Harvey follows + `@odata.nextLink` until Graph stops sending one, using the link exactly + as given, without re-adding the first request's parameters or touching + `$skip`, as Microsoft asks. Messages repeated across pages (new mail + shifts `$skip` mid-read) are returned once. +- **Links are checked before the token is attached.** A `nextLink` must be + `https://graph.microsoft.com` (default port), on the path + `/v1.0/me/mailFolders/inbox/messages`, with a query, and no credentials, + fragment, backslash or control characters. A link that repeats an earlier + page is a loop. Either way the read fails without sending the request. +- **Bounded.** More than `MAX_INBOX_PAGES` (40) pages, i.e. 2,000 messages + in a week at the default page size, fails the read rather than truncating. +- **Malformed data fails.** Any Graph response must be a JSON object; only + `202`/`204` (send, delete) may be empty. A page whose `value` is missing + or isn't a list, or a message with the wrong shape (no `id`, `from` or + `body` not an object, non-string fields, headers not a list) raises. A + message with no sender address is skipped, as before. + +Limitations: a mailbox with more than 2,000 inbox messages in 7 days holds +sending every cycle until the volume drops (raise `MAX_INBOX_PAGES` or the +page size). If Graph ever returns a `nextLink` in another shape (say +`/users/{id}/...`), reads fail closed until the check is widened. Mail that +arrives while pages are being read can shift `$skip` so one message is +missed that cycle; the next cycle reads it, since the window is 7 days. + +The loopback callback server listens on both `127.0.0.1` and `::1`, because +`localhost` resolves to both and browsers differ in which one they try. diff --git a/harvey.yaml b/harvey.yaml index 4bc60a1..8c2f94b 100644 --- a/harvey.yaml +++ b/harvey.yaml @@ -47,7 +47,7 @@ icp: channels: email: enabled: true - provider: "gmail" # gmail (recommended) | smtp | instantly + provider: "gmail" # gmail (recommended) | microsoft | smtp | instantly max_daily_sends: 50 require_approval: true # native providers: emails wait in the Outbox send_to_risky: false # also send to catch-all domains (off = safest) diff --git a/harvey/cli.py b/harvey/cli.py index 7fff8e5..4495867 100644 --- a/harvey/cli.py +++ b/harvey/cli.py @@ -392,6 +392,28 @@ async def _test(): asyncio.run(_test()) +def cmd_microsoft(args): + """Microsoft 365 provider utilities (auth / test).""" + from harvey.config import load_env + + env = load_env() + if args.microsoft_action == "auth": + from harvey.integrations.microsoft import run_auth_flow + ok = run_auth_flow(env.m365_client_id, env.m365_tenant_id) + sys.exit(0 if ok else 1) + + if args.microsoft_action == "test": + from harvey.config import load_config + from harvey.integrations.microsoft import MicrosoftProvider + + async def _test(): + provider = MicrosoftProvider(load_config(), env) + ok, detail = await provider.test_connection() + print(f"\n {'✓' if ok else '✗'} {detail}\n") + sys.exit(0 if ok else 1) + asyncio.run(_test()) + + def cmd_outbox(args): """Review and approve queued outgoing emails.""" from harvey.state import StateManager @@ -711,6 +733,12 @@ def main(): help="auth: one-time OAuth; test: verify connection") sub.set_defaults(func=cmd_gmail) + # harvey microsoft auth|test + sub = subparsers.add_parser("microsoft", help="Microsoft 365 provider setup") + sub.add_argument("microsoft_action", choices=["auth", "test"], + help="auth: one-time sign-in; test: verify connection") + sub.set_defaults(func=cmd_microsoft) + # harvey outbox sub = subparsers.add_parser("outbox", help="Review/approve queued emails") sub.add_argument("--approve", metavar="ID", default="", help="Approve one item") diff --git a/harvey/config.py b/harvey/config.py index 5095e70..36d866f 100644 --- a/harvey/config.py +++ b/harvey/config.py @@ -71,7 +71,8 @@ class ICPConfig(BaseModel): class EmailChannelConfig(BaseModel): enabled: bool = True # "instantly" (legacy), "gmail" (Gmail/Workspace via API — recommended), - # or "smtp" (any SMTP+IMAP mailbox: AgentMail, Fastmail, ...) + # "microsoft" (Microsoft 365 / Outlook via Graph), or "smtp" (any + # SMTP+IMAP mailbox: AgentMail, Fastmail, ...) provider: str = "instantly" max_daily_sends: int = 50 # When True, also send to catch-all ("risky") domains, not just verified @@ -289,6 +290,9 @@ class EnvConfig(BaseModel): imap_port: int = 993 imap_username: str = "" imap_password: str = "" + # Microsoft 365 (channels.email.provider: microsoft) + m365_client_id: str = "" + m365_tenant_id: str = "" # Optional: only used when compliance.precleared.enabled is true. precleared_api_key: str = "" @@ -401,6 +405,8 @@ def load_env() -> EnvConfig: imap_port=int(os.getenv("IMAP_PORT", "993").strip() or 993), imap_username=os.getenv("IMAP_USERNAME", "").strip(), imap_password=os.getenv("IMAP_PASSWORD", "").strip(), + m365_client_id=os.getenv("M365_CLIENT_ID", "").strip(), + m365_tenant_id=os.getenv("M365_TENANT_ID", "").strip(), precleared_api_key=os.getenv("PRECLEARED_API_KEY", "").strip(), ) return env diff --git a/harvey/dashboard.py b/harvey/dashboard.py index 054a141..fb16ef7 100644 --- a/harvey/dashboard.py +++ b/harvey/dashboard.py @@ -150,6 +150,11 @@ def _has(*keys): (PROJECT_ROOT / "data" / "gmail_token.json").is_file() provider_help = ("Set GMAIL_CLIENT_ID/SECRET in Settings, then run " "'harvey gmail auth' in your terminal.") + elif provider == "microsoft": + provider_done = _has("M365_CLIENT_ID") and \ + (PROJECT_ROOT / "data" / "m365_token.json").is_file() + provider_help = ("Set M365_CLIENT_ID and M365_TENANT_ID in .env, then run " + "'harvey microsoft auth' in your terminal.") elif provider == "smtp": provider_done = _has("SMTP_HOST", "SMTP_USERNAME", "SMTP_PASSWORD") provider_help = "Enter your SMTP host, username, and password in Settings." diff --git a/harvey/integrations/mail_provider.py b/harvey/integrations/mail_provider.py index 7813d57..24b3f7c 100644 --- a/harvey/integrations/mail_provider.py +++ b/harvey/integrations/mail_provider.py @@ -6,6 +6,8 @@ - ``gmail`` — a Google Workspace / Gmail mailbox via the Gmail REST API (recommended: a real mailbox on a dedicated secondary domain is the most deliverable way to send <50 cold emails/day). +- ``microsoft`` — a Microsoft 365 / Outlook mailbox via Microsoft Graph + (OAuth; Exchange Online no longer accepts passwords over IMAP). - ``smtp`` — any mailbox that speaks SMTP + IMAP (AgentMail, Fastmail, a Workspace app-mailbox, self-hosted). Configuration via .env. @@ -19,7 +21,7 @@ logger = logging.getLogger("harvey.mail") -NATIVE_PROVIDERS = ("gmail", "smtp") +NATIVE_PROVIDERS = ("gmail", "smtp", "microsoft") @dataclass @@ -96,4 +98,7 @@ def get_mail_provider(config, env) -> MailProvider | None: if provider == "smtp": from harvey.integrations.smtp_mail import SmtpImapProvider return SmtpImapProvider(config, env) + if provider == "microsoft": + from harvey.integrations.microsoft import MicrosoftProvider + return MicrosoftProvider(config, env) return None diff --git a/harvey/integrations/microsoft.py b/harvey/integrations/microsoft.py new file mode 100644 index 0000000..c958d82 --- /dev/null +++ b/harvey/integrations/microsoft.py @@ -0,0 +1,513 @@ +"""Microsoft 365 / Outlook mailbox provider — Microsoft Graph, no SDK. + +Microsoft has turned off password sign-in for IMAP in Exchange Online, so +the SMTP+IMAP provider can't read an M365 inbox. This one uses OAuth: + +Setup (one-time): +1. https://entra.microsoft.com → App registrations → New registration. + Name "Harvey", "Accounts in this organizational directory only". + Redirect URI: platform "Public client/native (mobile & desktop)", + value http://localhost:8766 +2. Copy the Application (client) ID and Directory (tenant) ID into .env as + M365_CLIENT_ID / M365_TENANT_ID. No client secret: Harvey is a public + client and signs in with PKCE. +3. Run ``harvey microsoft auth`` — opens a browser, you sign in and accept + Mail.ReadWrite + Mail.Send, and the refresh token is stored in + data/m365_token.json (mode 600). Revoke it any time at + https://myaccount.microsoft.com → Apps. + +Harvey never marks your mail read or moves it: it reads the last 7 days of +the inbox and remembers what it has handled. Mail is sent as the signed-in +mailbox, under its own display name. +""" + +import asyncio +import base64 +import hashlib +import http.server +import json +import logging +import secrets +import socket +import time +import webbrowser +from datetime import datetime, timedelta, timezone + +import httpx + +from harvey.paths import PROJECT_ROOT +from harvey.integrations.mail_provider import ( + InboundMessage, + MailProvider, + SendResult, + looks_like_bounce, +) + +logger = logging.getLogger("harvey.microsoft") + +TOKEN_FILE = PROJECT_ROOT / "data" / "m365_token.json" +GRAPH = "https://graph.microsoft.com/v1.0" +GRAPH_HOST = "graph.microsoft.com" +INBOX_MESSAGES = "/me/mailFolders/inbox/messages" +INBOX_MESSAGES_PATH = "/v1.0" + INBOX_MESSAGES +# Inbox pages read per cycle (limit messages each). More than this and the +# read fails, and sending is held, rather than acting on a partial inbox. +MAX_INBOX_PAGES = 40 +LOGIN = "https://login.microsoftonline.com" +SCOPE = ( + "offline_access https://graph.microsoft.com/Mail.ReadWrite " + "https://graph.microsoft.com/Mail.Send https://graph.microsoft.com/User.Read" +) +LOOPBACK_PORT = 8766 +REPLY_WINDOW_DAYS = 7 +MAX_RETRY_AFTER_SECONDS = 10 + +TIMEOUT = httpx.Timeout(connect=10.0, read=30.0, write=30.0, pool=10.0) + +_SELECT = ",".join(( + "id", "subject", "from", "body", "internetMessageId", "conversationId", + "receivedDateTime", "internetMessageHeaders", +)) + + +class GraphError(RuntimeError): + pass + + +def _load_token() -> dict | None: + try: + if TOKEN_FILE.is_file(): + data = json.loads(TOKEN_FILE.read_text()) + if isinstance(data, dict) and data.get("refresh_token"): + return data + except (OSError, json.JSONDecodeError) as e: + logger.warning(f"Could not read Microsoft token file: {e}") + return None + + +def _save_token(data: dict): + TOKEN_FILE.parent.mkdir(parents=True, exist_ok=True) + TOKEN_FILE.write_text(json.dumps(data, indent=2)) + try: + TOKEN_FILE.chmod(0o600) + except OSError: + pass + + +def _token_url(tenant: str) -> str: + return f"{LOGIN}/{tenant or 'organizations'}/oauth2/v2.0/token" + + +class MicrosoftProvider(MailProvider): + name = "microsoft" + + def __init__(self, config, env, transport=None): + self.config = config + self.client_id = getattr(env, "m365_client_id", "") + self.tenant = getattr(env, "m365_tenant_id", "") or "organizations" + self._transport = transport # httpx.MockTransport in tests + self._access_token = "" + self._access_expires = 0.0 + self._me = "" + + def is_configured(self) -> bool: + return bool(self.client_id and _load_token()) + + def _client(self) -> httpx.AsyncClient: + return httpx.AsyncClient(timeout=TIMEOUT, transport=self._transport) + + async def _token(self) -> str: + """Valid access token. Microsoft rotates the refresh token on every + use, so the new one is saved each time.""" + if self._access_token and time.time() < self._access_expires - 60: + return self._access_token + stored = _load_token() + if not stored: + raise GraphError("not authorized. Run 'harvey microsoft auth'.") + try: + async with self._client() as client: + resp = await client.post(_token_url(self.tenant), data={ + "client_id": self.client_id, + "grant_type": "refresh_token", + "refresh_token": stored["refresh_token"], + "scope": SCOPE, + }) + except httpx.HTTPError as e: + raise GraphError(f"token refresh failed: {e}") from e + if resp.status_code != 200: + raise GraphError( + f"token refresh rejected ({resp.status_code}): {resp.text[:200]}. " + "Re-run 'harvey microsoft auth'." + ) + data = resp.json() + if data.get("refresh_token"): + _save_token({**stored, "refresh_token": data["refresh_token"], + "refreshed_at": int(time.time())}) + self._access_token = data.get("access_token", "") + self._access_expires = time.time() + float(data.get("expires_in", 3600)) + if not self._access_token: + raise GraphError("token refresh returned no access token") + return self._access_token + + async def _request(self, method: str, path: str, **kwargs) -> dict: + return await self._request_url(method, f"{GRAPH}{path}", **kwargs) + + async def _request_url(self, method: str, url: str, *, headers=None, **kwargs) -> dict: + """One Graph call. Raises GraphError on failure; retries once when + Graph asks to slow down (429/503 with a short Retry-After). The + result is always a JSON object; only 202/204 may be empty (send, + delete), and come back as {}.""" + path = url[len(GRAPH):] if url.startswith(GRAPH) else url + token = await self._token() + all_headers = {"Authorization": f"Bearer {token}", **(headers or {})} + for attempt in (1, 2): + try: + async with self._client() as client: + resp = await client.request( + method, url, headers=all_headers, **kwargs + ) + except httpx.HTTPError as e: + raise GraphError(f"{method} {path} failed: {e}") from e + if resp.status_code in (429, 503) and attempt == 1: + try: + wait = float(resp.headers.get("Retry-After", "5")) + except ValueError: + wait = 5.0 + if wait <= MAX_RETRY_AFTER_SECONDS: + await asyncio.sleep(wait) + continue + break + if resp.status_code >= 400: + raise GraphError(f"Graph {resp.status_code} on {method} {path}: {resp.text[:200]}") + if not resp.content: + if resp.status_code in (202, 204): + return {} + raise GraphError(f"Graph {resp.status_code} on {method} {path}: empty response") + try: + data = resp.json() + except ValueError as e: + raise GraphError(f"Graph {resp.status_code} on {method} {path}: malformed JSON") from e + if not isinstance(data, dict): + raise GraphError( + f"Graph {resp.status_code} on {method} {path}: expected a JSON object, " + f"got {type(data).__name__}" + ) + return data + + async def _my_address(self) -> str: + if not self._me: + me = await self._request("GET", "/me", params={"$select": "mail,userPrincipalName"}) + self._me = (me.get("mail") or me.get("userPrincipalName") or "").lower() + return self._me + + async def _check_identity(self) -> str: + """The signed-in mailbox must be persona.email: a token for the wrong + account (say, the admin who approved the app) would otherwise send + cold email from that account.""" + me = await self._my_address() + expected = (getattr(getattr(self.config, "persona", None), "email", "") or "").strip().lower() + if expected and me != expected: + raise GraphError( + f"signed in as {me}, but persona.email is {expected}. Run " + f"'harvey microsoft auth' again and sign in as {expected}." + ) + return me + + async def send_email( + self, + to_email: str, + subject: str, + body: str, + thread_ref: str = "", + in_reply_to: str = "", + ) -> SendResult: + """Draft, then send, so the RFC Message-ID is known (bounces and + replies are matched on it). A reply is drafted with createReply on + the prospect's message, which sets In-Reply-To/References.""" + content = { + "subject": subject, + "body": {"contentType": "Text", "content": body}, + "toRecipients": [{"emailAddress": {"address": to_email}}], + } + draft = None + try: + await self._check_identity() + if thread_ref: + try: + draft = await self._request( + "POST", f"/me/messages/{thread_ref}/createReply" + ) + await self._request("PATCH", f"/me/messages/{draft['id']}", json=content) + except GraphError as e: + # Their message is gone (deleted/moved): send unthreaded. + logger.warning(f"Microsoft: createReply failed ({e}); sending as new.") + if draft and draft.get("id"): + await self._delete_draft(draft["id"]) + draft = None + if draft is None: + draft = await self._request("POST", "/me/messages", json=content) + await self._request("POST", f"/me/messages/{draft['id']}/send") + except (GraphError, KeyError) as e: + if draft and draft.get("id"): + await self._delete_draft(draft["id"]) + logger.error(f"Microsoft send failed to {to_email}: {e}") + return SendResult(ok=False, error=str(e)[:300]) + return SendResult( + ok=True, + message_id=draft.get("internetMessageId", ""), + thread_ref=draft.get("conversationId", ""), + ) + + async def _delete_draft(self, draft_id: str): + try: + await self._request("DELETE", f"/me/messages/{draft_id}") + except GraphError as e: + logger.warning(f"Microsoft: couldn't delete unsent draft {draft_id}: {e}") + + async def get_replies(self, limit: int = 50) -> list[InboundMessage]: + """The last week of the inbox, newest first, minus your own mail. + Every page is read (``limit`` is only the page size), following + Graph's @odata.nextLink as-is. Raises on any failure, including a + later page: the heartbeat then holds sends, because an unread reply + can't have stopped its follow-ups. Never returns a partial inbox.""" + me = await self._check_identity() + since = (datetime.now(timezone.utc) - timedelta(days=REPLY_WINDOW_DAYS)) + url = str(httpx.URL(f"{GRAPH}{INBOX_MESSAGES}", params={ + "$filter": f"receivedDateTime ge {since.strftime('%Y-%m-%dT%H:%M:%SZ')}", + "$orderby": "receivedDateTime desc", + "$top": str(max(1, int(limit))), + "$select": _SELECT, + })) + seen_urls = {url} + seen_ids = set() + out = [] + for page in range(1, MAX_INBOX_PAGES + 1): + data = await self._request_url( + "GET", url, headers={"Prefer": 'outlook.body-content-type="text"'}, + ) + items = data.get("value") + if not isinstance(items, list): + raise GraphError(f"inbox page {page}: 'value' is missing or not a list") + for item in items: + msg = self._parse_message(item) + # A message can repeat across pages if mail arrives mid-read. + if msg and msg.from_email != me and msg.provider_id not in seen_ids: + seen_ids.add(msg.provider_id) + out.append(msg) + next_link = data.get("@odata.nextLink") + if next_link is None: + return out + url = _checked_next_link(next_link) + if url in seen_urls: + raise GraphError(f"inbox page {page}: @odata.nextLink repeats an earlier page") + seen_urls.add(url) + raise GraphError( + f"inbox has more than {MAX_INBOX_PAGES} pages; refusing a partial read" + ) + + @staticmethod + def _parse_message(item) -> InboundMessage | None: + """One Graph message. A message with no sender address is skipped; + one with fields of the wrong shape raises GraphError.""" + def obj(value, what) -> dict: + if value is None: + return {} + if not isinstance(value, dict): + raise GraphError(f"malformed inbox message: {what} is not an object") + return value + + def text(value, what) -> str: + if value is None: + return "" + if not isinstance(value, str): + raise GraphError(f"malformed inbox message: {what} is not a string") + return value + + if not isinstance(item, dict): + raise GraphError("malformed inbox message: not an object") + msg_id = text(item.get("id"), "id") + if not msg_id: + raise GraphError("malformed inbox message: no id") + sender = obj(obj(item.get("from"), "from").get("emailAddress"), "from.emailAddress") + from_email = text(sender.get("address"), "sender address").strip().lower() + if not from_email: + return None + subject = text(item.get("subject"), "subject") + raw_headers = item.get("internetMessageHeaders") or [] + if not isinstance(raw_headers, list): + raise GraphError("malformed inbox message: internetMessageHeaders is not a list") + headers = {} + for h in raw_headers: + h = obj(h, "header") + headers[text(h.get("name"), "header name").lower()] = text( + h.get("value"), "header value") + body = text(obj(item.get("body"), "body").get("content"), "body content") + return InboundMessage( + provider_id=msg_id, + from_email=from_email, + subject=subject, + body=body[:5000], + message_id=text(item.get("internetMessageId"), "internetMessageId"), + in_reply_to=headers.get("in-reply-to", ""), + # The Graph id: what createReply needs to answer this message. + thread_ref=msg_id, + date=text(item.get("receivedDateTime"), "receivedDateTime"), + is_bounce=looks_like_bounce(from_email, subject), + headers=headers, + ) + + async def test_connection(self) -> tuple[bool, str]: + if not self.client_id: + return False, "M365_CLIENT_ID missing from .env" + if not _load_token(): + return False, "Not authorized yet — run 'harvey microsoft auth'" + try: + me = await self._check_identity() + await self._request( + "GET", "/me/mailFolders/inbox/messages", + params={"$top": "1", "$select": "id"}, + ) + except GraphError as e: + return False, str(e)[:300] + return True, f"Connected as {me}" + + +def _checked_next_link(link) -> str: + """Graph's @odata.nextLink, used whole (Microsoft: don't rebuild or + edit $skip). It gets the bearer token, so it must point back at the + inbox listing on Graph itself: https, graph.microsoft.com, the same + path, no credentials, no fragment. Anything else raises GraphError.""" + if not isinstance(link, str) or not link: + raise GraphError("@odata.nextLink is not a URL") + if any(c in link for c in "#\\") or any(c.isspace() or ord(c) < 32 for c in link): + raise GraphError("@odata.nextLink rejected: unexpected characters") + try: + url = httpx.URL(link) + except (httpx.InvalidURL, ValueError) as e: + raise GraphError(f"@odata.nextLink rejected: {e}") from e + path = url.raw_path.split(b"?", 1)[0].decode("ascii", "replace") + if ( + url.scheme != "https" + or url.host != GRAPH_HOST + or url.port not in (None, 443) + or url.userinfo + or url.fragment + or path.lower() != INBOX_MESSAGES_PATH.lower() + or not url.query + ): + raise GraphError("@odata.nextLink rejected: not the Graph inbox listing") + return link + + +def _pkce_pair() -> tuple[str, str]: + verifier = secrets.token_urlsafe(64)[:96] + challenge = base64.urlsafe_b64encode( + hashlib.sha256(verifier.encode()).digest() + ).decode().rstrip("=") + return verifier, challenge + + +class _LoopbackServer(http.server.HTTPServer): + allow_reuse_address = True + + +class _LoopbackServerV6(_LoopbackServer): + address_family = socket.AF_INET6 + + +def run_auth_flow(client_id: str, tenant: str = "") -> bool: + """One-time sign-in (authorization code + PKCE, loopback redirect). + Blocks until the browser round-trip completes.""" + import threading + import urllib.parse + + if not client_id: + print(" Set M365_CLIENT_ID (and M365_TENANT_ID) in .env first.") + print(" (entra.microsoft.com → App registrations → New registration)") + return False + tenant = tenant or "organizations" + + state = secrets.token_urlsafe(16) + verifier, challenge = _pkce_pair() + redirect_uri = f"http://localhost:{LOOPBACK_PORT}" + auth_url = f"{LOGIN}/{tenant}/oauth2/v2.0/authorize?" + urllib.parse.urlencode({ + "client_id": client_id, + "response_type": "code", + "redirect_uri": redirect_uri, + "response_mode": "query", + "scope": SCOPE, + "state": state, + "code_challenge": challenge, + "code_challenge_method": "S256", + "prompt": "select_account", + }) + + result: dict = {} + done = threading.Event() + + class Handler(http.server.BaseHTTPRequestHandler): + def do_GET(self): + query = urllib.parse.parse_qs(urllib.parse.urlparse(self.path).query) + if query.get("state", [""])[0] != state: + self.send_response(400); self.end_headers() + return + result["code"] = query.get("code", [""])[0] + result["error"] = (query.get("error_description") or query.get("error") or [""])[0] + self.send_response(200) + self.send_header("Content-Type", "text/html") + self.end_headers() + self.wfile.write( + b"

Harvey is connected to Microsoft 365.

You can close this tab.") + done.set() + + def log_message(self, *args): + pass + + # "localhost" is both 127.0.0.1 and ::1, and browsers differ in which + # they try: listen on both (loopback only), or the redirect can land on + # the one nobody is listening to. + servers = [_LoopbackServer(("127.0.0.1", LOOPBACK_PORT), Handler)] + try: + servers.append(_LoopbackServerV6(("::1", LOOPBACK_PORT), Handler)) + except OSError: + pass # no IPv6 loopback on this machine + for server in servers: + threading.Thread(target=server.serve_forever, daemon=True).start() + + print(f"\n Opening browser for Microsoft sign-in...\n {auth_url}\n") + webbrowser.open(auth_url) + finished = done.wait(timeout=300) + for server in servers: + server.shutdown() + server.server_close() + if not finished: + print(" Timed out waiting for the browser round-trip.") + return False + + if result.get("error") or not result.get("code"): + print(f" Authorization failed: {result.get('error') or 'no code returned'}") + return False + + resp = httpx.post(_token_url(tenant), data={ + "client_id": client_id, + "grant_type": "authorization_code", + "code": result["code"], + "redirect_uri": redirect_uri, + "code_verifier": verifier, + "scope": SCOPE, + }, timeout=30) + if resp.status_code != 200: + print(f" Token exchange failed ({resp.status_code}): {resp.text[:300]}") + return False + data = resp.json() + if not data.get("refresh_token"): + print(" Microsoft returned no refresh token (offline_access missing?).") + return False + _save_token({ + "refresh_token": data["refresh_token"], + "tenant": tenant, + "scope": data.get("scope", SCOPE), + "obtained_at": int(time.time()), + }) + print(f" ✓ Microsoft 365 authorized. Token stored at {TOKEN_FILE}") + return True diff --git a/tests/test_microsoft.py b/tests/test_microsoft.py new file mode 100644 index 0000000..1c92667 --- /dev/null +++ b/tests/test_microsoft.py @@ -0,0 +1,544 @@ +"""Microsoft 365 provider: token rotation, draft-then-send, threaded replies, +reading the inbox without touching it, and failing loudly.""" + +import json +import os + +import httpx +import pytest + +from harvey.integrations import microsoft +from harvey.integrations.microsoft import GraphError, MicrosoftProvider + + +class Env: + m365_client_id = "client-123" + m365_tenant_id = "tenant-abc" + + +class Graph: + """A tiny fake of the Graph + login endpoints Harvey uses.""" + + def __init__(self): + self.requests = [] + self.inbox = [] + self.fail = {} # (method, path suffix) -> status + self.drafts = 0 + self.pages = {} # inbox page number -> httpx.Response, or a dict + # merged into that page's normal JSON body + + def __call__(self, request: httpx.Request): + self.requests.append(request) + path = request.url.path + method = request.method + for (m, suffix), status in self.fail.items(): + if m == method and path.endswith(suffix): + return httpx.Response(status, json={"error": {"code": "boom"}}) + if request.url.host == "login.microsoftonline.com": + return httpx.Response(200, json={ + "access_token": "at-1", "expires_in": 3600, "refresh_token": "rt-2", + }) + if path == "/v1.0/me": + return httpx.Response(200, json={"mail": "Owner@sender.example"}) + if path == "/v1.0/me/messages" and method == "POST": + self.drafts += 1 + return httpx.Response(201, json={ + "id": f"draft{self.drafts}", + "internetMessageId": f"", + "conversationId": "conv1", + }) + if path.endswith("/createReply"): + self.drafts += 1 + return httpx.Response(201, json={ + "id": f"reply{self.drafts}", + "internetMessageId": f"", + "conversationId": "conv9", + }) + if method == "PATCH" or method == "DELETE": + return httpx.Response(200 if method == "PATCH" else 204, json={} if method == "PATCH" else None) + if path.endswith("/send"): + return httpx.Response(202) + if path == "/v1.0/me/mailFolders/inbox/messages": + return self._inbox_page(request) + return httpx.Response(404, json={"error": {"code": "notFound"}}) + + def _inbox_page(self, request): + """Pages like Graph: $top per page, then an opaque nextLink whose + query carries only $skip (and a token) — not the first page's params.""" + params = request.url.params + top = int(params.get("$top") or params.get("t") or 50) + skip = int(params.get("$skip", 0)) + page = skip // top + 1 + body = {"value": self.inbox[skip:skip + top]} + if skip + top < len(self.inbox): + body["@odata.nextLink"] = ( + "https://graph.microsoft.com/v1.0/me/mailFolders/inbox/messages" + f"?%24skip={skip + top}&t={top}&%24skiptoken=opaque{page}" + ) + override = self.pages.get(page) + if isinstance(override, httpx.Response): + return override + return httpx.Response(200, json={**body, **(override or {})}) + + def calls(self): + return [(r.method, r.url.path) for r in self.requests + if r.url.host == "graph.microsoft.com"] + + def inbox_requests(self): + return [r for r in self.requests if r.url.path.endswith("/inbox/messages")] + + +def message(i, sender="someone@elsewhere.example", body="hello"): + return {"id": f"m{i}", "subject": f"note {i}", + "from": {"emailAddress": {"address": sender}}, + "body": {"content": body}, "internetMessageId": f"<{i}@elsewhere.example>", + "receivedDateTime": "2026-10-14T15:00:00Z"} + + +@pytest.fixture +def graph(tmp_path, monkeypatch): + token_file = tmp_path / "m365_token.json" + token_file.write_text(json.dumps({"refresh_token": "rt-1", "tenant": "tenant-abc"})) + monkeypatch.setattr(microsoft, "TOKEN_FILE", token_file) + return Graph() + + +def provider(graph): + return MicrosoftProvider(config=None, env=Env(), transport=httpx.MockTransport(graph)) + + +@pytest.mark.asyncio +async def test_refresh_token_rotates_and_is_saved(graph): + p = provider(graph) + await p._token() + + [login] = [r for r in graph.requests if r.url.host == "login.microsoftonline.com"] + assert login.url.path == "/tenant-abc/oauth2/v2.0/token" + form = dict(x.split("=", 1) for x in login.content.decode().split("&")) + assert form["grant_type"] == "refresh_token" and form["refresh_token"] == "rt-1" + assert "client_secret" not in form # public client + saved = json.loads(microsoft.TOKEN_FILE.read_text()) + assert saved["refresh_token"] == "rt-2" + if os.name == "posix": # Windows has no POSIX mode bits; ACLs apply + assert microsoft.TOKEN_FILE.stat().st_mode & 0o777 == 0o600 + + +@pytest.mark.asyncio +async def test_new_email_is_drafted_then_sent_as_plain_text(graph): + result = await provider(graph).send_email("jane@acme.com", "booking", "Hi Jane.") + + assert result.ok + assert result.message_id == "" # bounces match on this + assert graph.calls() == [ + ("GET", "/v1.0/me"), # which mailbox the token belongs to + ("POST", "/v1.0/me/messages"), + ("POST", "/v1.0/me/messages/draft1/send"), + ] + [draft] = [json.loads(r.content) for r in graph.requests + if r.method == "POST" and r.url.path == "/v1.0/me/messages"] + assert draft["body"] == {"contentType": "Text", "content": "Hi Jane."} + assert draft["toRecipients"] == [{"emailAddress": {"address": "jane@acme.com"}}] + + +@pytest.mark.asyncio +async def test_reply_is_threaded_with_create_reply(graph): + result = await provider(graph).send_email( + "jane@acme.com", "Re: booking", "Tuesday works.", + thread_ref="AAMk-their-message", in_reply_to="", + ) + + assert result.ok and result.message_id == "" + assert graph.calls() == [ + ("GET", "/v1.0/me"), + ("POST", "/v1.0/me/messages/AAMk-their-message/createReply"), + ("PATCH", "/v1.0/me/messages/reply1"), + ("POST", "/v1.0/me/messages/reply1/send"), + ] + [patch] = [json.loads(r.content) for r in graph.requests if r.method == "PATCH"] + assert patch["body"]["content"] == "Tuesday works." + assert patch["toRecipients"][0]["emailAddress"]["address"] == "jane@acme.com" + + +@pytest.mark.asyncio +async def test_reply_to_a_deleted_message_is_sent_unthreaded(graph): + graph.fail[("POST", "/createReply")] = 404 + result = await provider(graph).send_email( + "jane@acme.com", "Re: booking", "Tuesday works.", thread_ref="gone", + ) + + assert result.ok + assert ("POST", "/v1.0/me/messages") in graph.calls() + + +@pytest.mark.asyncio +async def test_failed_send_deletes_the_draft(graph): + graph.fail[("POST", "/send")] = 500 + result = await provider(graph).send_email("jane@acme.com", "booking", "Hi.") + + assert not result.ok + assert ("DELETE", "/v1.0/me/messages/draft1") in graph.calls() + + +@pytest.mark.asyncio +async def test_inbox_is_read_without_changing_it(graph): + graph.inbox = [ + {"id": "m1", "subject": "Re: booking", + "from": {"emailAddress": {"address": "Jane@Acme.com"}}, + "body": {"contentType": "text", "content": "Sounds good."}, + "internetMessageId": "", "conversationId": "c1", + "receivedDateTime": "2026-10-14T15:00:00Z", + "internetMessageHeaders": [{"name": "In-Reply-To", "value": ""}]}, + {"id": "m2", "subject": "Undeliverable: booking", + "from": {"emailAddress": {"address": "postmaster@sender.example"}}, + "body": {"content": "couldn't be delivered"}, + "internetMessageHeaders": [{"name": "In-Reply-To", "value": ""}]}, + {"id": "m3", "subject": "note to self", + "from": {"emailAddress": {"address": "owner@sender.example"}}, "body": {"content": "x"}}, + ] + replies = await provider(graph).get_replies() + + assert [m.provider_id for m in replies] == ["m1", "m2"] # own mail skipped + jane, bounce = replies + assert jane.from_email == "jane@acme.com" + assert jane.body == "Sounds good." + assert jane.thread_ref == "m1" # what createReply needs + assert jane.in_reply_to == "" + assert bounce.is_bounce and bounce.in_reply_to == "" + + listing = [r for r in graph.requests if r.url.path.endswith("/inbox/messages")][0] + assert listing.headers["Prefer"] == 'outlook.body-content-type="text"' + assert "receivedDateTime ge" in listing.url.params["$filter"] + assert all(m in ("GET",) for m, _ in graph.calls()) # nothing marked read + + +@pytest.mark.asyncio +async def test_unreadable_inbox_raises_so_sends_are_held(graph): + graph.fail[("GET", "/inbox/messages")] = 503 + with pytest.raises(GraphError): + await provider(graph).get_replies() + + +@pytest.mark.asyncio +async def test_not_authorized_raises(graph): + microsoft.TOKEN_FILE.unlink() + p = provider(graph) + assert not p.is_configured() + with pytest.raises(GraphError, match="harvey microsoft auth"): + await p.get_replies() + + +@pytest.mark.asyncio +async def test_handler_holds_sends_when_graph_is_down(graph, tmp_path): + """End to end with the handler: a Graph outage is a failed read.""" + import os + from harvey.agents.handler import Handler + from harvey.state import StateManager + from tests.test_outbox_native import Cfg, StubBrain + + class E(Env): + instantly_api_key = "" + + state = StateManager(os.path.join(tmp_path, "t.db")) + await state.init_db() + handler = Handler(brain=StubBrain(), state=state, config=Cfg(), env=E()) + handler.provider = provider(graph) + graph.fail[("GET", "/inbox/messages")] = 500 + + assert await handler.run() is False + + +def test_registry_builds_the_microsoft_provider(): + from harvey.integrations.mail_provider import NATIVE_PROVIDERS, get_mail_provider + from tests.test_outbox_native import Cfg + + class MsCfg(Cfg): + class channels(Cfg.channels): + class email(Cfg.channels.email): + provider = "microsoft" + + assert "microsoft" in NATIVE_PROVIDERS + assert isinstance(get_mail_provider(MsCfg(), Env()), MicrosoftProvider) + + +@pytest.mark.asyncio +async def test_wrong_account_is_refused(graph): + class C: + class persona: + email = "you@sender.example" + + p = MicrosoftProvider(config=C(), env=Env(), transport=httpx.MockTransport(graph)) + # The fake Graph says the token belongs to owner@sender.example + result = await p.send_email("jane@acme.com", "booking", "Hi.") + assert not result.ok and "you@sender.example" in result.error + assert ("POST", "/v1.0/me/messages") not in graph.calls() + with pytest.raises(GraphError, match="signed in as owner@sender.example"): + await p.get_replies() + ok, detail = await p.test_connection() + assert not ok and "persona.email" in detail + + +def test_pkce_challenge_matches_verifier(): + import base64 + import hashlib + + verifier, challenge = microsoft._pkce_pair() + assert 43 <= len(verifier) <= 128 + expected = base64.urlsafe_b64encode(hashlib.sha256(verifier.encode()).digest()) + assert challenge == expected.decode().rstrip("=") + + +# ── Pagination and malformed responses: never a partial inbox ── + + +@pytest.mark.asyncio +async def test_every_page_of_the_week_is_read(graph): + graph.inbox = [message(i) for i in range(120)] + replies = await provider(graph).get_replies() + + assert [m.provider_id for m in replies] == [f"m{i}" for i in range(120)] + first, *later = graph.inbox_requests() + assert len(later) == 2 + assert first.url.params["$top"] == "50" + for request in later: + # The nextLink is followed exactly, with no initial params re-added, + # and still gets the same Prefer header. + assert "$filter" not in request.url.params and "$top" not in request.url.params + assert request.headers["Prefer"] == 'outlook.body-content-type="text"' + assert str(later[0].url) == ( + "https://graph.microsoft.com/v1.0/me/mailFolders/inbox/messages" + "?%24skip=50&t=50&%24skiptoken=opaque1" + ) + + +@pytest.mark.asyncio +async def test_limit_is_only_the_page_size(graph): + graph.inbox = [message(i) for i in range(7)] + replies = await provider(graph).get_replies(limit=3) + + assert len(replies) == 7 + assert len(graph.inbox_requests()) == 3 + + +@pytest.mark.asyncio +async def test_empty_inbox_is_an_empty_list(graph): + assert await provider(graph).get_replies() == [] + assert len(graph.inbox_requests()) == 1 + + +@pytest.mark.asyncio +@pytest.mark.parametrize("response", [ + httpx.Response(200, content=b"not json"), + httpx.Response(200, content=b'{"value": ['), + httpx.Response(200, json=[{"id": "m1"}]), + httpx.Response(200, json="value"), + httpx.Response(200, json={}), + httpx.Response(200, json={"value": None}), + httpx.Response(200, json={"value": {"id": "m1"}}), + httpx.Response(200), +], ids=["html", "truncated", "list", "string", "no-value", "null-value", + "object-value", "empty-200"]) +async def test_malformed_first_page_raises(graph, response): + graph.pages[1] = response + with pytest.raises(GraphError): + await provider(graph).get_replies() + + +@pytest.mark.asyncio +@pytest.mark.parametrize("override", [ + httpx.Response(500, json={"error": {"code": "boom"}}), + httpx.Response(200, content=b"not json"), + httpx.Response(200, json=["m51"]), + {"value": None}, + {"value": [message(50), "not a message"]}, + {"value": [{**message(50), "id": None}]}, + {"value": [{**message(50), "from": "jane@acme.com"}]}, + {"value": [{**message(50), "from": {"emailAddress": {"address": 7}}}]}, + {"value": [{**message(50), "body": "text"}]}, + {"value": [{**message(50), "internetMessageHeaders": {"In-Reply-To": "x"}}]}, + {"value": [{**message(50), "internetMessageHeaders": [{"name": 1}]}]}, + {"@odata.nextLink": 42}, + {"@odata.nextLink": ""}, +], ids=["http-500", "not-json", "list", "null-value", "non-object-message", + "no-id", "from-string", "address-number", "body-string", "headers-object", + "header-name-number", "nextlink-number", "nextlink-empty"]) +async def test_a_bad_later_page_fails_the_whole_read(graph, override): + graph.inbox = [message(i) for i in range(120)] + graph.pages[2] = override + with pytest.raises(GraphError): + await provider(graph).get_replies() + + +@pytest.mark.asyncio +@pytest.mark.parametrize("link", [ + "http://graph.microsoft.com/v1.0/me/mailFolders/inbox/messages?%24skip=50", + "https://evil.example/v1.0/me/mailFolders/inbox/messages?%24skip=50", + "https://graph.microsoft.com.evil.example/v1.0/me/mailFolders/inbox/messages?%24skip=50", + "https://graph.microsoft.com@evil.example/v1.0/me/mailFolders/inbox/messages?%24skip=50", + "https://user:pass@graph.microsoft.com/v1.0/me/mailFolders/inbox/messages?%24skip=50", + "https://graph.microsoft.com:8443/v1.0/me/mailFolders/inbox/messages?%24skip=50", + "https://graph.microsoft.com/v1.0/me/mailFolders/inbox/messages?%24skip=50#frag", + "https://graph.microsoft.com/v1.0/me/mailFolders/inbox/messages?%24skip=50#", + "https://graph.microsoft.com\\@evil.example/v1.0/me/mailFolders/inbox/messages?x=1", + "https://graph.microsoft.com/v1.0/me/messages?%24skip=50", + "https://graph.microsoft.com/v1.0/me/mailFolders/inbox/messages/m1/send?x=1", + "https://graph.microsoft.com/v1.0/me/mailFolders/inbox/messages/../../../sendMail?x=1", + "https://graph.microsoft.com/v1.0/me/mailFolders/inbox%2Fmessages?x=1", + "https://graph.microsoft.com/beta/me/mailFolders/inbox/messages?%24skip=50", + "https://graph.microsoft.com/v1.0/me/mailFolders/inbox/messages", + "/v1.0/me/mailFolders/inbox/messages?%24skip=50", + "https://graph.microsoft.com/v1.0/me/mailFolders/inbox/messages?a=1\r\nX-Evil: 1", +], ids=["http", "off-origin", "suffix-host", "userinfo-host", "credentials", "port", + "fragment", "empty-fragment", "backslash", "other-path", "send-path", + "dot-segments", "encoded-slash", "beta", "no-query", "relative", "crlf"]) +async def test_hostile_next_link_is_refused_before_any_request(graph, link): + graph.inbox = [message(i) for i in range(60)] + graph.pages[1] = {"@odata.nextLink": link} + with pytest.raises(GraphError, match="nextLink"): + await provider(graph).get_replies() + + # Only the first page went out; nothing (and no token) followed the link. + assert len(graph.inbox_requests()) == 1 + assert all(r.url.host in ("graph.microsoft.com", "login.microsoftonline.com") + for r in graph.requests) + graph_requests = [r for r in graph.requests if r.url.host == "graph.microsoft.com"] + assert [r.url.path for r in graph_requests] == [ + "/v1.0/me", "/v1.0/me/mailFolders/inbox/messages"] + + +@pytest.mark.asyncio +async def test_next_link_loop_is_refused(graph): + graph.inbox = [message(i) for i in range(120)] + loop = "https://graph.microsoft.com/v1.0/me/mailFolders/inbox/messages?%24skip=50&t=50" + graph.pages[1] = {"@odata.nextLink": loop} + graph.pages[2] = {"@odata.nextLink": loop} + with pytest.raises(GraphError, match="repeats"): + await provider(graph).get_replies() + assert len(graph.inbox_requests()) == 2 + + +@pytest.mark.asyncio +async def test_next_link_back_to_the_first_page_is_refused(graph): + def self_link(request): + if request.url.path.endswith("/inbox/messages"): + graph.requests.append(request) + return httpx.Response(200, json={ + "value": [message(0)], "@odata.nextLink": str(request.url)}) + return graph(request) + + p = MicrosoftProvider(config=None, env=Env(), transport=httpx.MockTransport(self_link)) + with pytest.raises(GraphError, match="repeats"): + await p.get_replies() + assert len(graph.inbox_requests()) == 1 + + +@pytest.mark.asyncio +async def test_too_many_pages_fails_closed(graph, monkeypatch): + monkeypatch.setattr(microsoft, "MAX_INBOX_PAGES", 3) + graph.inbox = [message(i) for i in range(10)] + with pytest.raises(GraphError, match="more than 3 pages"): + await provider(graph).get_replies(limit=3) + assert len(graph.inbox_requests()) == 3 + + graph.requests.clear() + graph.inbox = graph.inbox[:9] # exactly 3 pages is fine + assert len(await provider(graph).get_replies(limit=3)) == 9 + + +@pytest.mark.asyncio +async def test_message_repeated_across_pages_is_returned_once(graph): + graph.inbox = [message(i) for i in range(60)] + graph.inbox[50] = message(49) # shifted by new mail arriving mid-read + replies = await provider(graph).get_replies() + assert [m.provider_id for m in replies].count("m49") == 1 + assert len(replies) == 59 + + +@pytest.mark.asyncio +async def test_message_without_a_sender_is_skipped(graph): + graph.inbox = [{**message(0), "from": None}, message(1)] + assert [m.provider_id for m in await provider(graph).get_replies()] == ["m1"] + + +# ── Empty responses: fine for send/delete, not for anything with a body ── + + +@pytest.mark.asyncio +async def test_empty_202_and_204_are_success(graph): + p = provider(graph) + assert await p._request("POST", "/me/messages/draft1/send") == {} + assert await p._request("DELETE", "/me/messages/draft1") == {} + + +@pytest.mark.asyncio +async def test_malformed_draft_response_fails_the_send(graph): + def bad_draft(request): + if request.method == "POST" and request.url.path == "/v1.0/me/messages": + graph.requests.append(request) + return httpx.Response(201, content=b"oops") + return graph(request) + + p = MicrosoftProvider(config=None, env=Env(), transport=httpx.MockTransport(bad_draft)) + result = await p.send_email("jane@acme.com", "booking", "Hi.") + assert not result.ok and "malformed JSON" in result.error + assert not any(path.endswith("/send") for _, path in graph.calls()) + + +@pytest.mark.asyncio +async def test_malformed_identity_response_raises(graph): + def not_json(request): + if request.url.path == "/v1.0/me": + return httpx.Response(200, content=b"nope") + return graph(request) + + p = MicrosoftProvider(config=None, env=Env(), transport=httpx.MockTransport(not_json)) + with pytest.raises(GraphError, match="malformed JSON"): + await p.get_replies() + + +# ── With the handler: later pages count, and a bad later page holds sends ── + + +async def _handler_with_prospect(tmp_path, graph): + import os + from harvey.state import StateManager + from tests.test_outbox_native import _now_iso, make_handler, seed_prospect + + state = StateManager(os.path.join(tmp_path, "t.db")) + await state.init_db() + pid = await seed_prospect(state, email="jane@acme.com", status="contacted") + await state.add_outbox_item( + prospect_id=pid, to_email="jane@acme.com", subject="s2", body="b", + send_at=_now_iso(), status="approved", campaign_id="c1", step=2, + ) + return state, pid, make_handler(state, provider(graph)) + + +@pytest.mark.asyncio +async def test_unsubscribe_on_a_later_page_is_honored(graph, tmp_path): + graph.inbox = [message(i) for i in range(60)] + graph.inbox[55] = message(55, sender="Jane@Acme.com", body="Please unsubscribe me.") + state, pid, handler = await _handler_with_prospect(tmp_path, graph) + + assert await handler.run() is True + assert len(graph.inbox_requests()) == 2 + assert (await state.get_prospect(pid)).status == "opted_out" + assert len(await state.get_outbox(status="cancelled")) == 1 + + +@pytest.mark.asyncio +@pytest.mark.parametrize("override", [ + httpx.Response(503, headers={"Retry-After": "60"}), + httpx.Response(200, content=b"not json"), + {"value": "nope"}, +], ids=["outage", "not-json", "bad-value"]) +async def test_bad_later_page_holds_sends_and_handles_nothing(graph, tmp_path, override): + graph.inbox = [message(i) for i in range(60)] + graph.inbox[3] = message(3, sender="jane@acme.com", body="Please unsubscribe me.") + graph.pages[2] = override + state, pid, handler = await _handler_with_prospect(tmp_path, graph) + + assert await handler.run() is False + # Page 1 was readable, but nothing from a partial inbox is acted on. + assert not await state.is_reply_processed("m3") + assert (await state.get_prospect(pid)).status == "contacted" + assert len(await state.get_outbox(status="approved")) == 1