Microsoft 365 account connected successfully"
+ f"{(' (' + m365_email + ')') if m365_email else ''}. "
+ "You may close this window.
"
+ ""
+ ""
+ )
+ return HTMLResponse(content=html)
+
+
+@router.post("/{source_id}/m365/callback", response_model=MailSourceResponse)
+async def m365_oauth_callback_post(
+ source_id: int,
+ payload: M365CallbackRequest,
+ db: Session = Depends(get_db),
+ _auth: dict = Depends(require_admin_auth),
+) -> MailSourceResponse:
+ """Exchange a Microsoft OAuth2 authorization code for Graph tokens."""
+ source = _get_source_or_404(source_id, db)
+
+ if source.method != "M365_GRAPH":
+ raise HTTPException(
+ status_code=status.HTTP_400_BAD_REQUEST,
+ detail="This endpoint is only available for M365_GRAPH sources.",
+ )
+
+ try:
+ token_data = MicrosoftGraphClient.exchange_code_for_tokens(
+ tenant_id=source.m365_tenant_id or "common",
+ client_id=source.m365_client_id or "",
+ client_secret=source.m365_client_secret or "",
+ code=payload.code,
+ redirect_uri=payload.redirect_uri,
+ )
+ except Exception as exc: # pylint: disable=broad-exception-caught
+ logger.error(
+ "Microsoft 365 token exchange error for source id=%d: %s",
+ int(source_id),
+ _redact_sensitive_text(exc),
+ )
+ raise HTTPException(
+ status_code=status.HTTP_400_BAD_REQUEST,
+ detail=(
+ "Token exchange failed. Please check the Microsoft 365 "
+ "connection settings and try again."
+ ),
+ ) from exc
+
+ access_token = token_data.get("access_token")
+ refresh_token = token_data.get("refresh_token")
+ if not access_token:
+ raise HTTPException(
+ status_code=status.HTTP_400_BAD_REQUEST,
+ detail="Microsoft did not return an access token.",
+ )
+
+ m365_email = MicrosoftGraphClient.get_account_email(access_token)
+ source.m365_access_token = access_token
+ if refresh_token:
+ source.m365_refresh_token = refresh_token
+ if m365_email:
+ source.m365_email = m365_email
+ source.updated_at = datetime.utcnow()
+ db.commit()
+ db.refresh(source)
+
+ logger.info(
+ "Microsoft 365 OAuth2 tokens saved for source id=%d (account=%s)",
+ int(source_id),
+ _sanitize_for_log(m365_email or "unknown"),
+ )
+ return _source_to_response(source)
+
+
+@router.post("/{source_id}/m365/fetch", response_model=Dict[str, Any])
+async def m365_fetch_reports(
+ source_id: int,
+ db: Session = Depends(get_db),
+ _auth: dict = Depends(require_admin_auth),
+) -> Dict[str, Any]:
+ """Manually trigger a Microsoft 365 Graph DMARC report fetch."""
+ source = _get_source_or_404(source_id, db)
+
+ if source.method != "M365_GRAPH":
+ raise HTTPException(
+ status_code=status.HTTP_400_BAD_REQUEST,
+ detail="This endpoint is only available for M365_GRAPH sources.",
+ )
+
+ results = _fetch_m365_source(source, db)
+ logger.info(
+ "Microsoft 365 fetch for source id=%d: processed=%d reports_found=%d",
+ int(source_id),
+ int(results.get("processed", 0)),
+ int(results.get("reports_found", 0)),
+ )
+ for err in results.get("errors", []):
+ logger.warning(
+ "Microsoft 365 fetch warning for source id=%d: %s",
+ int(source_id),
+ _redact_sensitive_text(err),
+ )
+
+ return _fetch_response(source, results) # lgtm[py/stack-trace-exposure]
+
+
+@router.delete("/{source_id}/m365/connection", status_code=status.HTTP_204_NO_CONTENT)
+async def m365_disconnect(
+ source_id: int,
+ db: Session = Depends(get_db),
+ _auth: dict = Depends(require_admin_auth),
+) -> None:
+ """Clear the stored Microsoft Graph OAuth2 tokens for this source."""
+ source = _get_source_or_404(source_id, db)
+
+ if source.method != "M365_GRAPH":
+ raise HTTPException(
+ status_code=status.HTTP_400_BAD_REQUEST,
+ detail="This endpoint is only available for M365_GRAPH sources.",
+ )
+
+ source.m365_access_token = None
+ source.m365_refresh_token = None
+ source.m365_email = None
+ source.updated_at = datetime.utcnow()
+ db.commit()
+ logger.info("Microsoft 365 tokens cleared for source id=%d", int(source_id))
+
+
# ---------------------------------------------------------------------------
# Gmail API OAuth2 routes
# ---------------------------------------------------------------------------
diff --git a/backend/app/main.py b/backend/app/main.py
index 4865236..36f5e23 100644
--- a/backend/app/main.py
+++ b/backend/app/main.py
@@ -30,6 +30,7 @@ from app.models.mail_source import MailSource # noqa: F401 – ensure table is
from app.services.gmail_client import GmailClient
from app.services.imap_client import IMAPClient
from app.services.import_history import record_import_attempt
+from app.services.microsoft_graph_client import MicrosoftGraphClient
from app.services.report_persistence import hydrate_report_store_from_db
from app.services.report_store import ReportStore
from app.services.runtime_status import (
@@ -162,7 +163,74 @@ def _poll_single_gmail_source(source: MailSource) -> None:
)
-def _poll_all_enabled_sources() -> list[MailSource]:
+def _poll_single_m365_source(source: MailSource) -> None:
+ """Fetch DMARC reports for a single M365_GRAPH mail source."""
+ global last_check_time # pylint: disable=global-statement
+
+ if not source.m365_access_token:
+ logger.info(
+ "Microsoft 365 polling (source id=%d): skipped – OAuth2 not yet authorised",
+ source.id,
+ )
+ return
+
+ db = SessionLocal()
+ try:
+ src = db.query(MailSource).get(source.id)
+ poll_source = src or source
+ already = MicrosoftGraphClient.load_ingested_ids(poll_source.m365_ingested_ids)
+ client = MicrosoftGraphClient(
+ tenant_id=poll_source.m365_tenant_id or "common",
+ client_id=poll_source.m365_client_id or "",
+ client_secret=poll_source.m365_client_secret or "",
+ access_token=poll_source.m365_access_token,
+ refresh_token=poll_source.m365_refresh_token or "",
+ mailbox=poll_source.m365_mailbox,
+ folder=poll_source.folder or "INBOX",
+ already_ingested_ids=already,
+ db=db,
+ )
+
+ started_at = datetime.utcnow()
+ results = client.fetch_reports()
+ if src:
+ if results.get("new_ingested_ids"):
+ all_ids = list(dict.fromkeys(already + results["new_ingested_ids"]))
+ src.m365_ingested_ids = MicrosoftGraphClient.dump_ingested_ids(all_ids)
+
+ refreshed = client.get_refreshed_tokens()
+ if refreshed:
+ src.m365_access_token = refreshed["access_token"]
+ if "refresh_token" in refreshed:
+ src.m365_refresh_token = refreshed["refresh_token"]
+
+ src.last_checked = datetime.utcnow()
+ record_import_attempt(db, src, results, started_at=started_at, trigger="scheduled")
+ db.commit()
+ finally:
+ db.close()
+
+ last_check_time = datetime.now()
+
+ if results["success"]:
+ logger.info(
+ "Microsoft 365 polling (source id=%d): %s emails processed, "
+ "%s aggregate reports found",
+ source.id,
+ results["processed"],
+ results["reports_found"],
+ )
+ if results["new_domains"]:
+ logger.info("New domains found: %s", ", ".join(results["new_domains"]))
+ else:
+ logger.error(
+ "Microsoft 365 polling (source id=%d) failed: %s",
+ source.id,
+ results.get("error", "Unknown error"),
+ )
+
+
+def _poll_all_enabled_sources() -> list[MailSource]: # noqa: C901
"""Iterate over all enabled mail sources and poll each one."""
db = SessionLocal()
try:
@@ -182,6 +250,11 @@ def _poll_all_enabled_sources() -> list[MailSource]:
_poll_single_gmail_source(source)
except Exception as e: # pylint: disable=broad-exception-caught
logger.error("Error polling Gmail source id=%d: %s", source.id, str(e))
+ elif source.method == "M365_GRAPH":
+ try:
+ _poll_single_m365_source(source)
+ except Exception as e: # pylint: disable=broad-exception-caught
+ logger.error("Error polling Microsoft 365 source id=%d: %s", source.id, str(e))
elif source.method == "IMAP":
try:
_poll_single_imap_source(source)
@@ -682,7 +755,50 @@ def _trigger_poll_gmail_source(source: MailSource, db) -> dict:
}
-def _poll_source_for_trigger(source: MailSource, db, days: int = 7) -> dict:
+def _trigger_poll_m365_source(source: MailSource, db) -> dict:
+ """Poll a single M365_GRAPH source and return a result dict for the API response."""
+ global last_check_time # pylint: disable=global-statement
+
+ already = MicrosoftGraphClient.load_ingested_ids(source.m365_ingested_ids)
+ graph_client = MicrosoftGraphClient(
+ tenant_id=source.m365_tenant_id or "common",
+ client_id=source.m365_client_id or "",
+ client_secret=source.m365_client_secret or "",
+ access_token=source.m365_access_token,
+ refresh_token=source.m365_refresh_token or "",
+ mailbox=source.m365_mailbox,
+ folder=source.folder or "INBOX",
+ already_ingested_ids=already,
+ db=db,
+ )
+ started_at = datetime.utcnow()
+ results = graph_client.fetch_reports()
+ last_check_time = datetime.now()
+
+ if results.get("new_ingested_ids"):
+ all_ids = list(dict.fromkeys(already + results["new_ingested_ids"]))
+ source.m365_ingested_ids = MicrosoftGraphClient.dump_ingested_ids(all_ids)
+ refreshed = graph_client.get_refreshed_tokens()
+ if refreshed:
+ source.m365_access_token = refreshed["access_token"]
+ if "refresh_token" in refreshed:
+ source.m365_refresh_token = refreshed["refresh_token"]
+ source.last_checked = datetime.utcnow()
+ record_import_attempt(db, source, results, started_at=started_at, trigger="manual")
+ db.commit()
+ return {
+ "source_id": source.id,
+ "name": source.name,
+ "success": results["success"],
+ "processed": results.get("processed", 0),
+ "reports_found": results.get("reports_found", 0),
+ "forensic_reports_found": results.get("forensic_reports_found", 0),
+ "duplicate_forensic_reports": results.get("duplicate_forensic_reports", 0),
+ "new_domains": results.get("new_domains", []),
+ }
+
+
+def _poll_source_for_trigger(source: MailSource, db, days: int = 7) -> dict: # noqa: C901
"""Dispatch a single mail source for the manual trigger-poll endpoint.
Returns a result/summary dict that is included in the API response.
@@ -705,6 +821,24 @@ def _poll_source_for_trigger(source: MailSource, db, days: int = 7) -> dict:
"success": False,
"error": "Failed to poll. Check server logs for details.",
}
+ if source.method == "M365_GRAPH":
+ if not source.m365_access_token:
+ return {
+ "source_id": source.id,
+ "name": source.name,
+ "skipped": True,
+ "reason": "Microsoft 365 account not yet authorised",
+ }
+ try:
+ return _trigger_poll_m365_source(source, db)
+ except Exception as e: # pylint: disable=broad-exception-caught
+ logger.error("Error polling Microsoft 365 source id=%d: %s", source.id, str(e))
+ return {
+ "source_id": source.id,
+ "name": source.name,
+ "success": False,
+ "error": "Failed to poll. Check server logs for details.",
+ }
if source.method == "IMAP":
try:
return _trigger_poll_imap_source(source, db, days=days)
diff --git a/backend/app/models/mail_source.py b/backend/app/models/mail_source.py
index 84d27cb..06376d7 100644
--- a/backend/app/models/mail_source.py
+++ b/backend/app/models/mail_source.py
@@ -18,6 +18,7 @@ class MailSource(Base):
- ``IMAP`` – standard IMAP4 (over SSL/TLS or STARTTLS)
- ``POP3`` – POP3 inbox (stub for future implementation)
- ``GMAIL_API`` – Gmail API with OAuth 2.0
+ - ``M365_GRAPH`` – Microsoft 365 / Exchange Online via Microsoft Graph
"""
__tablename__ = "mail_sources"
@@ -28,7 +29,7 @@ class MailSource(Base):
name = Column(String, nullable=False)
# Connection method – determines which fields are used at runtime
- method = Column(String, nullable=False, default="IMAP") # IMAP | POP3 | GMAIL_API
+ method = Column(String, nullable=False, default="IMAP") # IMAP | POP3 | GMAIL_API | M365_GRAPH
# Connection details (used by IMAP and POP3)
server = Column(String, nullable=True)
@@ -48,6 +49,19 @@ class MailSource(Base):
# JSON-encoded list of Gmail message IDs that have already been ingested
gmail_ingested_ids = Column(Text, nullable=True, default="[]")
+ # Microsoft 365 / Graph OAuth2 credentials (used by M365_GRAPH method)
+ m365_tenant_id = Column(String, nullable=True, default="common")
+ m365_client_id = Column(String, nullable=True)
+ _m365_client_secret = Column("m365_client_secret", Text, nullable=True)
+ _m365_access_token = Column("m365_access_token", Text, nullable=True)
+ _m365_refresh_token = Column("m365_refresh_token", Text, nullable=True)
+ # Optional user/shared mailbox to poll. Empty means the authorised account (/me).
+ m365_mailbox = Column(String, nullable=True)
+ # Email address reported by Microsoft Graph for the authorised account.
+ m365_email = Column(String, nullable=True)
+ # JSON-encoded list of Graph message IDs that have already been ingested
+ m365_ingested_ids = Column(Text, nullable=True, default="[]")
+
# Polling behaviour
polling_interval = Column(Integer, default=60) # minutes
@@ -76,6 +90,9 @@ class MailSource(Base):
"gmail_client_secret": self._gmail_client_secret,
"gmail_access_token": self._gmail_access_token,
"gmail_refresh_token": self._gmail_refresh_token,
+ "m365_client_secret": self._m365_client_secret,
+ "m365_access_token": self._m365_access_token,
+ "m365_refresh_token": self._m365_refresh_token,
}
for public_name, stored_value in secret_fields.items():
@@ -120,3 +137,30 @@ class MailSource(Base):
@gmail_refresh_token.setter
def gmail_refresh_token(self, value):
self._gmail_refresh_token = encrypt_secret(value)
+
+ @property
+ def m365_client_secret(self):
+ """Return the decrypted Microsoft 365 OAuth client secret, if present."""
+ return decrypt_secret(self._m365_client_secret)
+
+ @m365_client_secret.setter
+ def m365_client_secret(self, value):
+ self._m365_client_secret = encrypt_secret(value)
+
+ @property
+ def m365_access_token(self):
+ """Return the decrypted Microsoft Graph access token, if present."""
+ return decrypt_secret(self._m365_access_token)
+
+ @m365_access_token.setter
+ def m365_access_token(self, value):
+ self._m365_access_token = encrypt_secret(value)
+
+ @property
+ def m365_refresh_token(self):
+ """Return the decrypted Microsoft Graph refresh token, if present."""
+ return decrypt_secret(self._m365_refresh_token)
+
+ @m365_refresh_token.setter
+ def m365_refresh_token(self, value):
+ self._m365_refresh_token = encrypt_secret(value)
diff --git a/backend/app/services/microsoft_graph_client.py b/backend/app/services/microsoft_graph_client.py
new file mode 100644
index 0000000..eb98773
--- /dev/null
+++ b/backend/app/services/microsoft_graph_client.py
@@ -0,0 +1,466 @@
+"""Microsoft Graph client for retrieving DMARC aggregate reports."""
+
+import base64
+import json
+import logging
+from typing import Any, Dict, List, Optional
+from urllib.parse import quote, urlencode
+
+import httpx
+
+from app.services.dmarc_parser import DMARCParser
+from app.services.report_persistence import report_exists, save_parsed_report
+from app.services.report_store import ReportStore
+
+logger = logging.getLogger(__name__)
+
+GRAPH_BASE_URL = "https://graph.microsoft.com/v1.0"
+LOGIN_BASE_URL = "https://login.microsoftonline.com"
+
+M365_SCOPES = [
+ "offline_access",
+ "https://graph.microsoft.com/User.Read",
+ "https://graph.microsoft.com/Mail.Read",
+]
+
+_PAGE_SIZE = 100
+_DMARC_SUBJECT_TERMS = (
+ "dmarc",
+ "aggregate report",
+ "domain report",
+ "report domain",
+ "rua",
+ "submitter",
+)
+_DMARC_SENDER_TERMS = (
+ "dmarc",
+ "reports",
+ "postmaster",
+)
+
+
+class MicrosoftGraphError(RuntimeError):
+ """Raised when Microsoft Graph or the token endpoint returns a failure."""
+
+
+class MicrosoftGraphClient:
+ """
+ Retrieve DMARC aggregate reports from Microsoft 365 through Microsoft Graph.
+
+ The client uses delegated OAuth tokens and read-only Graph scopes. Messages
+ are never modified or deleted; already-ingested Graph message IDs are stored
+ by the caller to avoid reprocessing the same email.
+ """
+
+ def __init__(
+ self,
+ tenant_id: str,
+ client_id: str,
+ client_secret: str,
+ access_token: str,
+ refresh_token: str,
+ mailbox: Optional[str] = None,
+ folder: str = "inbox",
+ already_ingested_ids: Optional[List[str]] = None,
+ db: Any = None,
+ ):
+ self.tenant_id = tenant_id or "common"
+ self.client_id = client_id
+ self.client_secret = client_secret
+ self.access_token = access_token
+ self.refresh_token = refresh_token
+ self.mailbox = (mailbox or "").strip()
+ self.folder = folder or "inbox"
+ self.already_ingested_ids: List[str] = list(already_ingested_ids or [])
+ self.report_store = ReportStore.get_instance()
+ self.db = db
+ self._refreshed_tokens: Optional[Dict[str, str]] = None
+
+ def get_refreshed_tokens(self) -> Optional[Dict[str, str]]:
+ """Return refreshed OAuth tokens, if a request had to refresh them."""
+ return self._refreshed_tokens
+
+ @staticmethod
+ def build_authorization_url(
+ tenant_id: str,
+ client_id: str,
+ redirect_uri: str,
+ state: Optional[str] = None,
+ ) -> str:
+ """Build a Microsoft identity platform authorization-code URL."""
+ tenant = quote(tenant_id or "common", safe="")
+ params: Dict[str, str] = {
+ "client_id": client_id,
+ "response_type": "code",
+ "redirect_uri": redirect_uri,
+ "response_mode": "query",
+ "scope": " ".join(M365_SCOPES),
+ "prompt": "select_account",
+ }
+ if state:
+ params["state"] = state
+ return f"{LOGIN_BASE_URL}/{tenant}/oauth2/v2.0/authorize?" + urlencode(params)
+
+ @staticmethod
+ def exchange_code_for_tokens(
+ tenant_id: str,
+ client_id: str,
+ client_secret: str,
+ code: str,
+ redirect_uri: str,
+ ) -> Dict[str, Any]:
+ """Exchange an authorization code for Microsoft Graph tokens."""
+ data = {
+ "client_id": client_id,
+ "client_secret": client_secret,
+ "code": code,
+ "redirect_uri": redirect_uri,
+ "grant_type": "authorization_code",
+ "scope": " ".join(M365_SCOPES),
+ }
+ resp = httpx.post(MicrosoftGraphClient._token_url(tenant_id), data=data, timeout=30)
+ if resp.status_code != 200:
+ raise MicrosoftGraphError(
+ f"Microsoft token exchange failed ({resp.status_code}): {resp.text}"
+ )
+ return resp.json()
+
+ @staticmethod
+ def get_account_email(access_token: str) -> Optional[str]:
+ """Return the mailbox identity exposed by Graph /me for an access token."""
+ try:
+ resp = httpx.get(
+ f"{GRAPH_BASE_URL}/me",
+ headers={"Authorization": f"Bearer {access_token}"},
+ params={"$select": "mail,userPrincipalName"},
+ timeout=30,
+ )
+ if resp.status_code == 200:
+ profile = resp.json()
+ return profile.get("mail") or profile.get("userPrincipalName")
+ except Exception as exc: # pylint: disable=broad-exception-caught
+ logger.error("Failed to fetch Microsoft 365 account email: %s", exc)
+ return None
+
+ @staticmethod
+ def load_ingested_ids(json_text: Optional[str]) -> List[str]:
+ """Deserialize the m365_ingested_ids text column into a list."""
+ if not json_text:
+ return []
+ try:
+ decoded = json.loads(json_text)
+ except (json.JSONDecodeError, TypeError):
+ return []
+ return [str(item) for item in decoded] if isinstance(decoded, list) else []
+
+ @staticmethod
+ def dump_ingested_ids(ids: List[str]) -> str:
+ """Serialize Graph message IDs for database storage."""
+ return json.dumps(ids)
+
+ def test_connection(self) -> Dict[str, Any]:
+ """Verify that the saved delegated token can read the target mailbox."""
+ mailbox_path = self._mailbox_path()
+ data = self._request(
+ "GET",
+ f"{mailbox_path}/messages",
+ params={"$top": 1, "$select": "id"},
+ )
+ return {
+ "success": True,
+ "message_count": len(data.get("value", [])),
+ "diagnostic_detail": "Microsoft Graph mailbox read succeeded.",
+ }
+
+ def fetch_reports(self) -> Dict[str, Any]:
+ """Fetch and ingest DMARC report attachments from Microsoft Graph."""
+ stats: Dict[str, Any] = {
+ "success": True,
+ "processed": 0,
+ "reports_found": 0,
+ "forensic_reports_found": 0,
+ "duplicate_reports": 0,
+ "duplicate_forensic_reports": 0,
+ "new_domains": [],
+ "errors": [],
+ "new_ingested_ids": [],
+ "details": [],
+ }
+
+ try:
+ messages = self._list_dmarc_messages()
+ 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)]}
+
+ domains_before = set(self.report_store.get_domains())
+
+ for message in messages:
+ message_id = str(message.get("id") or "")
+ if not message_id:
+ continue
+ if message_id in self.already_ingested_ids:
+ self._append_detail(
+ stats,
+ status="skipped",
+ reason="already_ingested_message",
+ message_id=message_id,
+ )
+ continue
+
+ stats["processed"] += 1
+ found = self._process_message(message, stats)
+ if found >= 0:
+ stats["new_ingested_ids"].append(message_id)
+ self.already_ingested_ids.append(message_id)
+
+ domains_after = set(self.report_store.get_domains())
+ stats["new_domains"] = list(domains_after - domains_before)
+ return stats
+
+ @staticmethod
+ def _token_url(tenant_id: str) -> str:
+ tenant = quote(tenant_id or "common", safe="")
+ return f"{LOGIN_BASE_URL}/{tenant}/oauth2/v2.0/token"
+
+ @staticmethod
+ def _append_detail(stats: dict, **detail: str) -> None:
+ stats.setdefault("details", []).append(
+ {key: value for key, value in detail.items() if value}
+ )
+
+ def _mailbox_path(self) -> str:
+ if not self.mailbox or self.mailbox.lower() == "me":
+ return "/me"
+ return f"/users/{quote(self.mailbox, safe='')}"
+
+ def _messages_path(self) -> str:
+ mailbox_path = self._mailbox_path()
+ folder = (self.folder or "").strip()
+ if not folder:
+ return f"{mailbox_path}/messages"
+ if folder.upper() == "INBOX":
+ folder = "inbox"
+ return f"{mailbox_path}/mailFolders/{quote(folder, safe='')}/messages"
+
+ def _headers(self) -> Dict[str, str]:
+ return {"Authorization": f"Bearer {self.access_token}"}
+
+ def _request(
+ self,
+ method: str,
+ path_or_url: str,
+ *,
+ 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 = httpx.request(method, url, headers=self._headers(), params=params, timeout=30)
+ if resp.status_code < 200 or resp.status_code >= 300:
+ raise MicrosoftGraphError(self._format_error(resp))
+ return resp.json() if resp.content else {}
+
+ def _refresh_access_token(self) -> None:
+ data = {
+ "client_id": self.client_id,
+ "client_secret": self.client_secret,
+ "refresh_token": self.refresh_token,
+ "grant_type": "refresh_token",
+ "scope": " ".join(M365_SCOPES),
+ }
+ resp = httpx.post(self._token_url(self.tenant_id), data=data, timeout=30)
+ if resp.status_code != 200:
+ raise MicrosoftGraphError(
+ f"Microsoft token refresh failed ({resp.status_code}): {resp.text}"
+ )
+ token_data = resp.json()
+ access_token = token_data.get("access_token")
+ if not access_token:
+ raise MicrosoftGraphError("Microsoft token refresh did not return an access token.")
+ self.access_token = access_token
+ refreshed = {"access_token": access_token}
+ if token_data.get("refresh_token"):
+ self.refresh_token = token_data["refresh_token"]
+ refreshed["refresh_token"] = token_data["refresh_token"]
+ self._refreshed_tokens = refreshed
+
+ @staticmethod
+ def _format_error(resp: httpx.Response) -> str:
+ try:
+ payload = resp.json()
+ except ValueError:
+ payload = {}
+ message = payload.get("error_description")
+ if not message and isinstance(payload.get("error"), dict):
+ message = payload["error"].get("message")
+ code = payload["error"].get("code")
+ if code:
+ message = f"{code}: {message}" if message else code
+ return message or f"Microsoft Graph request failed ({resp.status_code}): {resp.text}"
+
+ @staticmethod
+ def _looks_like_dmarc_message(message: Dict[str, Any]) -> bool:
+ if not message.get("hasAttachments"):
+ return False
+ subject = str(message.get("subject") or "").lower()
+ sender = (
+ ((message.get("from") or {}).get("emailAddress") or {}).get("address") or ""
+ ).lower()
+ return any(term in subject for term in _DMARC_SUBJECT_TERMS) or any(
+ term in sender for term in _DMARC_SENDER_TERMS
+ )
+
+ @staticmethod
+ def _is_dmarc_attachment(filename: str) -> bool:
+ lower = filename.lower()
+ return (
+ lower.endswith(".xml")
+ or lower.endswith(".zip")
+ or lower.endswith(".gz")
+ or lower.endswith(".gzip")
+ )
+
+ def _list_dmarc_messages(self) -> List[Dict[str, Any]]:
+ messages: List[Dict[str, Any]] = []
+ url = self._messages_path()
+ params: Optional[Dict[str, Any]] = {
+ "$top": _PAGE_SIZE,
+ "$select": "id,subject,from,hasAttachments,receivedDateTime",
+ "$orderby": "receivedDateTime desc",
+ }
+
+ while url:
+ data = self._request("GET", url, params=params)
+ for message in data.get("value", []):
+ if self._looks_like_dmarc_message(message):
+ messages.append(message)
+ url = data.get("@odata.nextLink")
+ params = None
+
+ return messages
+
+ def _process_message(self, message: Dict[str, Any], stats: Dict[str, Any]) -> int:
+ message_id = str(message.get("id") or "")
+ try:
+ attachments = self._list_attachments(message_id)
+ except Exception as exc: # pylint: disable=broad-exception-caught
+ logger.error("Microsoft Graph: failed to fetch attachments for %s: %s", message_id, exc)
+ stats["errors"].append(f"Failed to fetch attachments for message {message_id}: {exc}")
+ self._append_detail(
+ stats,
+ status="error",
+ reason="attachment_fetch_failed",
+ message_id=message_id,
+ error=str(exc),
+ )
+ return -1
+ return self._process_attachments(message_id, attachments, stats)
+
+ def _list_attachments(self, message_id: str) -> List[Dict[str, Any]]:
+ mailbox_path = self._mailbox_path()
+ data = self._request(
+ "GET",
+ f"{mailbox_path}/messages/{quote(message_id, safe='')}/attachments",
+ )
+ return list(data.get("value", []))
+
+ def _store_report_if_new(self, report: Dict[str, Any]) -> bool:
+ domain = report.get("domain", "unknown")
+ report_id = report.get("report_id", "")
+ if report_id and (
+ self.report_store.has_report(domain, report_id)
+ or (self.db is not None and report_exists(self.db, domain, report_id))
+ ):
+ logger.info("Skipping duplicate DMARC report %s for %s", report_id, domain)
+ return False
+
+ if self.db is not None:
+ save_parsed_report(self.db, report)
+ self.report_store.add_report(report)
+ return True
+
+ def _process_attachments(
+ self,
+ message_id: str,
+ attachments: List[Dict[str, Any]],
+ stats: Dict[str, Any],
+ ) -> int:
+ reports_found = 0
+
+ for attachment in attachments:
+ filename = str(attachment.get("name") or "")
+ if not filename:
+ continue
+ if not self._is_dmarc_attachment(filename):
+ self._append_detail(
+ stats,
+ status="skipped",
+ reason="unsupported_attachment",
+ message_id=message_id,
+ filename=filename,
+ )
+ continue
+
+ attachment_type = str(attachment.get("@odata.type") or "").lower()
+ if "fileattachment" not in attachment_type:
+ self._append_detail(
+ stats,
+ status="skipped",
+ reason="unsupported_attachment_type",
+ message_id=message_id,
+ filename=filename,
+ )
+ continue
+
+ content_b64 = attachment.get("contentBytes")
+ if not content_b64:
+ self._append_detail(
+ stats,
+ status="skipped",
+ reason="empty_attachment",
+ message_id=message_id,
+ filename=filename,
+ )
+ continue
+
+ try:
+ content = base64.b64decode(content_b64)
+ report = DMARCParser.parse_file(content, filename)
+ domain = str(report.get("domain", "unknown"))
+ report_id = str(report.get("report_id", ""))
+ if self._store_report_if_new(report):
+ stats["reports_found"] += 1
+ reports_found += 1
+ self._append_detail(
+ stats,
+ status="imported",
+ message_id=message_id,
+ filename=filename,
+ domain=domain,
+ report_id=report_id,
+ )
+ else:
+ stats["duplicate_reports"] = stats.get("duplicate_reports", 0) + 1
+ self._append_detail(
+ stats,
+ status="duplicate",
+ message_id=message_id,
+ filename=filename,
+ domain=domain,
+ report_id=report_id,
+ )
+ except Exception as exc: # pylint: disable=broad-exception-caught
+ logger.error("Failed to parse Graph DMARC attachment %s: %s", filename, exc)
+ stats["errors"].append(f"Failed to parse {filename}: {exc}")
+ self._append_detail(
+ stats,
+ status="error",
+ reason="parse_failed",
+ message_id=message_id,
+ filename=filename,
+ error=str(exc),
+ )
+
+ return reports_found
diff --git a/backend/app/templates/mail_sources.html b/backend/app/templates/mail_sources.html
index aba217a..df091da 100644
--- a/backend/app/templates/mail_sources.html
+++ b/backend/app/templates/mail_sources.html
@@ -86,8 +86,8 @@
-
-
+
+
IMAP
+
@@ -434,6 +435,85 @@
+
+
+
+
+
+
Microsoft 365 Setup
+
Enter an app registration Client ID and client secret from Microsoft Entra admin center. Add this app's callback URL as a Web redirect URI, then save and click Connect Microsoft 365.
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+ Connected as
+
+
+
+
+ Not yet authorised. Save this source first, then click Connect Microsoft 365.
+