From 8556ea30bfe7b503e2d6341eafd6dc51ad95a276 Mon Sep 17 00:00:00 2001 From: Christian Krakau-Louis Date: Sat, 23 May 2026 15:42:52 +0200 Subject: [PATCH] feat: add m365 safe backfill windows --- .../app/api/api_v1/endpoints/mail_sources.py | 12 +++-- backend/app/main.py | 10 ++-- .../app/services/microsoft_graph_client.py | 53 ++++++++++++++++--- backend/app/tests/test_mail_sources.py | 12 +++-- backend/app/tests/test_main_polling.py | 1 + .../app/tests/test_microsoft_graph_client.py | 51 ++++++++++++++++++ docs/milestones.md | 2 +- docs/user_guide/microsoft365.md | 4 +- 8 files changed, 122 insertions(+), 23 deletions(-) diff --git a/backend/app/api/api_v1/endpoints/mail_sources.py b/backend/app/api/api_v1/endpoints/mail_sources.py index 0490bf9..4a3ed15 100644 --- a/backend/app/api/api_v1/endpoints/mail_sources.py +++ b/backend/app/api/api_v1/endpoints/mail_sources.py @@ -12,7 +12,7 @@ import logging from datetime import datetime from typing import Any, Dict, List, Optional -from fastapi import APIRouter, Depends, HTTPException, Request, status +from fastapi import APIRouter, Depends, HTTPException, Query, Request, status from pydantic import BaseModel from sqlalchemy.orm import Session @@ -446,6 +446,7 @@ def _fetch_response(source: MailSource, results: Dict[str, Any]) -> Dict[str, An "error_count": len(results.get("errors", [])), "target_mailbox": results.get("target_mailbox"), "target_folder": results.get("target_folder"), + "search_window_days": results.get("search_window_days"), "timestamp": datetime.now().isoformat(), } @@ -487,7 +488,7 @@ def _fetch_gmail_source(source: MailSource, db: Session) -> Dict[str, Any]: return results -def _fetch_m365_source(source: MailSource, db: Session) -> Dict[str, Any]: +def _fetch_m365_source(source: MailSource, db: Session, days: int = 7) -> Dict[str, Any]: """Run one Microsoft 365 Graph import and persist source/import metadata.""" if not source.m365_access_token: raise HTTPException( @@ -510,7 +511,7 @@ def _fetch_m365_source(source: MailSource, db: Session) -> Dict[str, Any]: ) started_at = datetime.utcnow() - results = client.fetch_reports() + results = client.fetch_reports(days=days) if results.get("new_ingested_ids"): all_ids = list(dict.fromkeys(already + results["new_ingested_ids"])) @@ -551,7 +552,7 @@ def _fetch_source(source: MailSource, db: Session, days: int) -> Dict[str, Any]: if source.method == "GMAIL_API": return _fetch_gmail_source(source, db) if source.method == "M365_GRAPH": - return _fetch_m365_source(source, db) + return _fetch_m365_source(source, db, days=days) if source.method == "IMAP": return _fetch_imap_source(source, db, days) raise HTTPException( @@ -1113,6 +1114,7 @@ async def m365_oauth_callback_post( @router.post("/{source_id}/m365/fetch", response_model=Dict[str, Any]) async def m365_fetch_reports( source_id: int, + days: int = Query(7, ge=1, le=365, title="Number of days to fetch"), db: Session = Depends(get_db), _auth: dict = Depends(require_admin_auth), ) -> Dict[str, Any]: @@ -1125,7 +1127,7 @@ async def m365_fetch_reports( detail="This endpoint is only available for M365_GRAPH sources.", ) - results = _fetch_m365_source(source, db) + results = _fetch_m365_source(source, db, days=days) logger.info( "Microsoft 365 fetch for source id=%d: processed=%d reports_found=%d", int(source_id), diff --git a/backend/app/main.py b/backend/app/main.py index e777d8b..74b22dc 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -193,7 +193,7 @@ def _poll_single_m365_source(source: MailSource) -> None: ) started_at = datetime.utcnow() - results = client.fetch_reports() + results = client.fetch_reports(days=7) if src: if results.get("new_ingested_ids"): all_ids = list(dict.fromkeys(already + results["new_ingested_ids"])) @@ -756,7 +756,7 @@ def _trigger_poll_gmail_source(source: MailSource, db) -> dict: } -def _trigger_poll_m365_source(source: MailSource, db) -> dict: +def _trigger_poll_m365_source(source: MailSource, db, days: int = 7) -> dict: """Poll a single M365_GRAPH source and return a result dict for the API response.""" global last_check_time # pylint: disable=global-statement @@ -774,7 +774,7 @@ def _trigger_poll_m365_source(source: MailSource, db) -> dict: db=db, ) started_at = datetime.utcnow() - results = graph_client.fetch_reports() + results = graph_client.fetch_reports(days=days) last_check_time = datetime.now() if results.get("new_ingested_ids"): @@ -832,7 +832,7 @@ def _poll_source_for_trigger(source: MailSource, db, days: int = 7) -> dict: # "reason": "Microsoft 365 account not yet authorised", } try: - return _trigger_poll_m365_source(source, db) + return _trigger_poll_m365_source(source, db, days=days) except Exception as e: # pylint: disable=broad-exception-caught logger.error("Error polling Microsoft 365 source id=%d: %s", source.id, str(e)) return { @@ -881,7 +881,7 @@ def _poll_enabled_sources_for_trigger(days: int) -> list[dict]: @app.post("/api/v1/admin/trigger-poll") async def trigger_imap_poll( auth: dict = Depends(require_admin_auth), - days: int = Query(7, ge=1, le=365, title="Number of days to fetch for IMAP sources"), + days: int = Query(7, ge=1, le=365, title="Number of days to fetch for mail sources"), ): """ Manually trigger IMAP polling for all enabled mail sources (admin only). diff --git a/backend/app/services/microsoft_graph_client.py b/backend/app/services/microsoft_graph_client.py index 942442b..9801e56 100644 --- a/backend/app/services/microsoft_graph_client.py +++ b/backend/app/services/microsoft_graph_client.py @@ -3,7 +3,9 @@ import base64 import json import logging -from typing import Any, Dict, List, Optional +import time +from datetime import datetime, timedelta +from typing import Any, Callable, Dict, List, Optional from urllib.parse import quote, urlencode import httpx @@ -25,6 +27,9 @@ M365_SCOPES = [ _PAGE_SIZE = 100 _MAX_FOLDER_DEPTH = 5 +_MAX_GRAPH_RETRIES = 3 +_MAX_RETRY_DELAY_SECONDS = 30 +_RETRYABLE_STATUS_CODES = {429, 503, 504} _DMARC_SUBJECT_TERMS = ( "dmarc", "aggregate report", @@ -65,6 +70,7 @@ class MicrosoftGraphClient: folder_id: Optional[str] = None, already_ingested_ids: Optional[List[str]] = None, db: Any = None, + sleep: Optional[Callable[[float], None]] = None, ): self.tenant_id = tenant_id or "common" self.client_id = client_id @@ -77,6 +83,7 @@ class MicrosoftGraphClient: self.already_ingested_ids: List[str] = list(already_ingested_ids or []) self.report_store = ReportStore.get_instance() self.db = db + self._sleep = sleep or time.sleep self._refreshed_tokens: Optional[Dict[str, str]] = None def get_refreshed_tokens(self) -> Optional[Dict[str, str]]: @@ -227,8 +234,9 @@ class MicrosoftGraphClient: url = data.get("@odata.nextLink") params = None - def fetch_reports(self) -> Dict[str, Any]: + def fetch_reports(self, days: int = 7) -> Dict[str, Any]: """Fetch and ingest DMARC report attachments from Microsoft Graph.""" + safe_days = max(1, min(int(days or 7), 365)) stats: Dict[str, Any] = { "success": True, "processed": 0, @@ -242,10 +250,11 @@ class MicrosoftGraphClient: "details": [], "target_mailbox": self._target_mailbox_label(), "target_folder": self._target_folder_label(), + "search_window_days": safe_days, } try: - messages = self._list_dmarc_messages() + messages = self._list_dmarc_messages(days=safe_days) except Exception as exc: # pylint: disable=broad-exception-caught logger.error("Microsoft Graph: failed to list messages: %s", exc) return {**stats, "success": False, "error": str(exc), "errors": [str(exc)]} @@ -324,14 +333,42 @@ class MicrosoftGraphClient: params: Optional[Dict[str, Any]] = None, ) -> Dict[str, Any]: url = path_or_url if path_or_url.startswith("http") else f"{GRAPH_BASE_URL}{path_or_url}" - resp = httpx.request(method, url, headers=self._headers(), params=params, timeout=30) - if resp.status_code == 401 and self.refresh_token: - self._refresh_access_token() + resp: Optional[httpx.Response] = None + + for attempt in range(_MAX_GRAPH_RETRIES + 1): resp = httpx.request(method, url, headers=self._headers(), params=params, timeout=30) + if resp.status_code == 401 and self.refresh_token: + self._refresh_access_token() + resp = httpx.request(method, url, headers=self._headers(), params=params, timeout=30) + if ( + resp.status_code in _RETRYABLE_STATUS_CODES + and attempt < _MAX_GRAPH_RETRIES + ): + delay = self._retry_delay_seconds(resp, attempt) + logger.warning( + "Microsoft Graph request throttled/unavailable; retrying in %.1fs", + delay, + ) + self._sleep(delay) + continue + break + + if resp is None: + raise MicrosoftGraphError("Microsoft Graph request failed before receiving a response.") if resp.status_code < 200 or resp.status_code >= 300: raise MicrosoftGraphError(self._format_error(resp)) return resp.json() if resp.content else {} + @staticmethod + def _retry_delay_seconds(resp: httpx.Response, attempt: int) -> float: + retry_after = resp.headers.get("Retry-After") + if retry_after: + try: + return min(float(retry_after), _MAX_RETRY_DELAY_SECONDS) + except ValueError: + pass + return min(float(2**attempt), _MAX_RETRY_DELAY_SECONDS) + def _refresh_access_token(self) -> None: data = { "client_id": self.client_id, @@ -392,13 +429,15 @@ class MicrosoftGraphClient: or lower.endswith(".gzip") ) - def _list_dmarc_messages(self) -> List[Dict[str, Any]]: + def _list_dmarc_messages(self, days: int) -> List[Dict[str, Any]]: messages: List[Dict[str, Any]] = [] url = self._messages_path() + cutoff = datetime.utcnow() - timedelta(days=days) params: Optional[Dict[str, Any]] = { "$top": _PAGE_SIZE, "$select": "id,subject,from,hasAttachments,receivedDateTime", "$orderby": "receivedDateTime desc", + "$filter": f"receivedDateTime ge {cutoff.strftime('%Y-%m-%dT%H:%M:%SZ')}", } while url: diff --git a/backend/app/tests/test_mail_sources.py b/backend/app/tests/test_mail_sources.py index 25698d9..f1fbcff 100644 --- a/backend/app/tests/test_mail_sources.py +++ b/backend/app/tests/test_mail_sources.py @@ -1455,6 +1455,7 @@ class TestMicrosoft365GraphMailSource: "new_domains": ["example.com"], "errors": [], "new_ingested_ids": ["id1", "id2", "id3"], + "search_window_days": 30, } mock_client.get_refreshed_tokens.return_value = None @@ -1472,7 +1473,7 @@ class TestMicrosoft365GraphMailSource: return_value='["id1", "id2", "id3"]', ), ): - resp = authed_client.post(f"/api/v1/mail-sources/{source_id}/fetch") + resp = authed_client.post(f"/api/v1/mail-sources/{source_id}/fetch?days=30") assert resp.status_code == 200 data = resp.json() @@ -1483,6 +1484,8 @@ class TestMicrosoft365GraphMailSource: assert data["new_domains"] == ["example.com"] assert data["target_mailbox"] is None assert data["target_folder"] is None + assert data["search_window_days"] == 30 + mock_client.fetch_reports.assert_called_once_with(days=30) def test_m365_specific_fetch_persists_refreshed_tokens( self, authed_client: TestClient, db_session: Session @@ -2667,12 +2670,13 @@ class TestTriggerPollM365Source: patch("app.main.MicrosoftGraphClient.load_ingested_ids", return_value=[]), patch("app.main.MicrosoftGraphClient.dump_ingested_ids", return_value='["id1"]'), ): - result = _trigger_poll_m365_source(src, mock_db) + result = _trigger_poll_m365_source(src, mock_db, days=30) assert result["success"] is True assert result["source_id"] == 8 assert result["processed"] == 2 assert src.m365_ingested_ids == '["id1"]' + mock_gc.fetch_reports.assert_called_once_with(days=30) mock_db.commit.assert_called_once() def test_persists_refreshed_tokens(self): @@ -2776,10 +2780,10 @@ class TestPollSourceForTrigger: expected = {"source_id": 8, "name": "M365", "success": True} with patch("app.main._trigger_poll_m365_source", return_value=expected) as mock_fn: - result = _poll_source_for_trigger(src, MagicMock()) + result = _poll_source_for_trigger(src, MagicMock(), days=30) assert result is expected - mock_fn.assert_called_once() + assert mock_fn.call_args.kwargs["days"] == 30 def test_m365_exception_returns_failure_dict(self): from app.main import _poll_source_for_trigger diff --git a/backend/app/tests/test_main_polling.py b/backend/app/tests/test_main_polling.py index 09d08bb..3229b73 100644 --- a/backend/app/tests/test_main_polling.py +++ b/backend/app/tests/test_main_polling.py @@ -133,6 +133,7 @@ def test_poll_single_m365_source_persists_import_state(): assert db_source.m365_ingested_ids == '["message-1"]' assert db_source.m365_access_token == "new-tok" assert db_source.m365_refresh_token == "new-ref" + mock_client.fetch_reports.assert_called_once_with(days=7) db.commit.assert_called_once() db.close.assert_called_once() diff --git a/backend/app/tests/test_microsoft_graph_client.py b/backend/app/tests/test_microsoft_graph_client.py index d907fc2..c59de44 100644 --- a/backend/app/tests/test_microsoft_graph_client.py +++ b/backend/app/tests/test_microsoft_graph_client.py @@ -192,6 +192,8 @@ class TestMicrosoftGraphFetchReports: def test_fetch_reports_uses_selected_folder_id_and_records_context(self, monkeypatch): def fake_request(method, url, headers=None, params=None, timeout=None): if url.endswith("/users/shared%40example.com/mailFolders/folder-id/messages"): + assert "$filter" in params + assert params["$filter"].startswith("receivedDateTime ge ") return httpx.Response( 200, json={ @@ -226,6 +228,55 @@ class TestMicrosoftGraphFetchReports: assert result["processed"] == 1 assert result["target_mailbox"] == "shared@example.com" assert result["target_folder"] == "DMARC Reports" + assert result["search_window_days"] == 7 + + def test_fetch_reports_applies_requested_search_window(self, monkeypatch): + def fake_request(method, url, headers=None, params=None, timeout=None): + assert url.endswith("/me/mailFolders/inbox/messages") + assert params["$top"] == 100 + assert params["$select"] == "id,subject,from,hasAttachments,receivedDateTime" + assert params["$orderby"] == "receivedDateTime desc" + assert params["$filter"].startswith("receivedDateTime ge ") + return httpx.Response(200, json={"value": []}) + + monkeypatch.setattr("app.services.microsoft_graph_client.httpx.request", fake_request) + client = _make_client() + + result = client.fetch_reports(days=30) + + assert result["success"] is True + assert result["search_window_days"] == 30 + assert result["processed"] == 0 + + def test_request_retries_graph_throttling(self, monkeypatch): + responses = [ + httpx.Response( + 429, + headers={"Retry-After": "0"}, + json={"error": {"code": "TooManyRequests", "message": "slow down"}}, + ), + httpx.Response(200, json={"value": []}), + ] + sleeps = [] + + def fake_request(method, url, headers=None, params=None, timeout=None): + return responses.pop(0) + + monkeypatch.setattr("app.services.microsoft_graph_client.httpx.request", fake_request) + client = MicrosoftGraphClient( + tenant_id="organizations", + client_id="client-id", + client_secret="client-secret", + access_token="access-token", + refresh_token="refresh-token", + sleep=sleeps.append, + ) + + result = client.fetch_reports() + + assert result["success"] is True + assert sleeps == [0.0] + assert responses == [] def test_fetch_reports_imports_zip_attachment(self, monkeypatch, db_session): attachment_bytes = _zip_xml() diff --git a/docs/milestones.md b/docs/milestones.md index 2ea41bd..c8482e5 100644 --- a/docs/milestones.md +++ b/docs/milestones.md @@ -206,7 +206,7 @@ Planned: - Microsoft 365 mail source using OAuth (Graph) with least-privilege scopes. Delivered for delegated `User.Read`, `Mail.Read`, and `offline_access` with encrypted token storage, manual import, scheduled polling, UI setup, and operator docs. - Shared mailbox and folder selection support for DMARC report collection. Delivered with shared mailbox targeting, Microsoft Graph folder listing, folder-id based imports, UI selection, and mailbox/folder context in import history. - Import-history parity with existing sources (auditable attachment outcomes, duplicates, parse failures). Delivered for Microsoft 365 imports. -- Backfill support with safe throttling and progressive search windows. +- Backfill support with safe throttling and progressive search windows. Delivered with days-based Graph `receivedDateTime` filters, duplicate-safe reruns, and retry/backoff for throttled or temporarily unavailable Graph requests. - Secret handling mirrors existing guidance (no raw secrets in logs; 1Password-friendly). Exit criteria: diff --git a/docs/user_guide/microsoft365.md b/docs/user_guide/microsoft365.md index 3ea4b06..17a5a26 100644 --- a/docs/user_guide/microsoft365.md +++ b/docs/user_guide/microsoft365.md @@ -39,10 +39,12 @@ DMARQ uses delegated Microsoft Graph access. That means the authorised Microsoft ## Import Behavior -DMARQ reads recent messages in the configured folder, filters for messages that look like DMARC reports, downloads Graph `fileAttachment` items, and sends `.xml`, `.zip`, `.gz`, and `.gzip` attachments through the same parser and persistence path used by upload, IMAP, and Gmail imports. +DMARQ reads messages in the configured search window, filters for messages that look like DMARC reports, downloads Graph `fileAttachment` items, and sends `.xml`, `.zip`, `.gz`, and `.gzip` attachments through the same parser and persistence path used by upload, IMAP, and Gmail imports. Imported Graph message IDs are stored on the mail source so scheduled polling does not reprocess the same message. Import history records processed messages, imported reports, duplicates, parse failures, attachment-level details, and the mailbox/folder target used for the attempt. +Manual imports and backfills use the **days** value from the Mail Sources UI or trigger-poll endpoint. DMARQ passes that window to Microsoft Graph as a `receivedDateTime` filter, so large mailboxes can be searched without scanning unrelated older mail. If Graph returns a throttling or temporary service response, DMARQ retries with `Retry-After` or exponential backoff before surfacing the sanitized error in import history. + ## Troubleshooting - **Not authorised**: reconnect the source from Mail Sources.