From 5caefb06dbd26d38d9eabf9bb2cf996f54ada613 Mon Sep 17 00:00:00 2001 From: Christian Krakau-Louis Date: Sat, 23 May 2026 14:49:54 +0200 Subject: [PATCH] feat: add Microsoft 365 Graph mail source --- ...6d7e8_add_m365_graph_mail_source_fields.py | 39 ++ .../app/api/api_v1/endpoints/mail_sources.py | 395 ++++++++++- backend/app/main.py | 138 +++- backend/app/models/mail_source.py | 46 +- .../app/services/microsoft_graph_client.py | 466 +++++++++++++ backend/app/templates/mail_sources.html | 174 ++++- backend/app/tests/test_mail_sources.py | 633 ++++++++++++++++++ backend/app/tests/test_main_polling.py | 62 ++ .../app/tests/test_microsoft_graph_client.py | 397 +++++++++++ docs/index.md | 4 +- docs/milestones.md | 6 +- docs/user_guide/microsoft365.md | 40 ++ 12 files changed, 2379 insertions(+), 21 deletions(-) create mode 100644 backend/alembic/versions/f3a4b5c6d7e8_add_m365_graph_mail_source_fields.py create mode 100644 backend/app/services/microsoft_graph_client.py create mode 100644 backend/app/tests/test_microsoft_graph_client.py create mode 100644 docs/user_guide/microsoft365.md diff --git a/backend/alembic/versions/f3a4b5c6d7e8_add_m365_graph_mail_source_fields.py b/backend/alembic/versions/f3a4b5c6d7e8_add_m365_graph_mail_source_fields.py new file mode 100644 index 0000000..038c827 --- /dev/null +++ b/backend/alembic/versions/f3a4b5c6d7e8_add_m365_graph_mail_source_fields.py @@ -0,0 +1,39 @@ +"""Add Microsoft 365 Graph mail source fields. + +Revision ID: f3a4b5c6d7e8 +Revises: e2f3a4b5c6d7 +Create Date: 2026-05-23 00:00:00.000000 +""" + +from typing import Sequence, Union + +import sqlalchemy as sa +from alembic import op + +# revision identifiers, used by Alembic. +revision: str = "f3a4b5c6d7e8" +down_revision: Union[str, Sequence[str], None] = "e2f3a4b5c6d7" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + op.add_column("mail_sources", sa.Column("m365_tenant_id", sa.String(), nullable=True)) + op.add_column("mail_sources", sa.Column("m365_client_id", sa.String(), nullable=True)) + op.add_column("mail_sources", sa.Column("m365_client_secret", sa.Text(), nullable=True)) + op.add_column("mail_sources", sa.Column("m365_access_token", sa.Text(), nullable=True)) + op.add_column("mail_sources", sa.Column("m365_refresh_token", sa.Text(), nullable=True)) + op.add_column("mail_sources", sa.Column("m365_mailbox", sa.String(), nullable=True)) + op.add_column("mail_sources", sa.Column("m365_email", sa.String(), nullable=True)) + op.add_column("mail_sources", sa.Column("m365_ingested_ids", sa.Text(), nullable=True)) + + +def downgrade() -> None: + op.drop_column("mail_sources", "m365_ingested_ids") + op.drop_column("mail_sources", "m365_email") + op.drop_column("mail_sources", "m365_mailbox") + op.drop_column("mail_sources", "m365_refresh_token") + op.drop_column("mail_sources", "m365_access_token") + op.drop_column("mail_sources", "m365_client_secret") + op.drop_column("mail_sources", "m365_client_id") + op.drop_column("mail_sources", "m365_tenant_id") diff --git a/backend/app/api/api_v1/endpoints/mail_sources.py b/backend/app/api/api_v1/endpoints/mail_sources.py index 602700b..5a3ea3b 100644 --- a/backend/app/api/api_v1/endpoints/mail_sources.py +++ b/backend/app/api/api_v1/endpoints/mail_sources.py @@ -3,8 +3,8 @@ Mail Sources API endpoints. Provides CRUD operations for MailSource objects stored in the database, plus a *test-connection* action that validates the supplied credentials without -persisting anything. Gmail API sources additionally have OAuth2 helper -endpoints (authorize-url, callback, fetch). +persisting anything. Gmail API and Microsoft 365 sources additionally have +OAuth2 helper endpoints (authorize-url, callback, fetch). """ import json @@ -24,6 +24,7 @@ from app.models.mail_source_import import MailSourceImport 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 router = APIRouter() logger = logging.getLogger(__name__) @@ -38,7 +39,7 @@ class MailSourceBase(BaseModel): """Fields shared by create and update payloads.""" name: str - method: str = "IMAP" # IMAP | POP3 | GMAIL_API + method: str = "IMAP" # IMAP | POP3 | GMAIL_API | M365_GRAPH server: Optional[str] = None port: int = 993 username: Optional[str] = None @@ -50,6 +51,11 @@ class MailSourceBase(BaseModel): # Gmail API OAuth2 fields (only relevant when method == GMAIL_API) gmail_client_id: Optional[str] = None gmail_client_secret: Optional[str] = None + # Microsoft 365 Graph OAuth2 fields (only relevant when method == M365_GRAPH) + m365_tenant_id: Optional[str] = "common" + m365_client_id: Optional[str] = None + m365_client_secret: Optional[str] = None + m365_mailbox: Optional[str] = None class MailSourceCreate(MailSourceBase): @@ -71,6 +77,10 @@ class MailSourceUpdate(BaseModel): enabled: Optional[bool] = None gmail_client_id: Optional[str] = None gmail_client_secret: Optional[str] = None + m365_tenant_id: Optional[str] = None + m365_client_id: Optional[str] = None + m365_client_secret: Optional[str] = None + m365_mailbox: Optional[str] = None class MailSourceResponse(MailSourceBase): @@ -86,6 +96,9 @@ class MailSourceResponse(MailSourceBase): gmail_email: Optional[str] = None # Indicate whether OAuth tokens are present (without exposing them) gmail_connected: bool = False + # Microsoft 365: show the authorised account and token state, but not tokens + m365_email: Optional[str] = None + m365_connected: bool = False class Config: from_attributes = True @@ -109,6 +122,13 @@ class GmailCallbackRequest(BaseModel): redirect_uri: str +class M365CallbackRequest(BaseModel): + """Payload for the Microsoft 365 OAuth2 callback endpoint.""" + + code: str + redirect_uri: str + + class MailSourceImportResponse(BaseModel): """Sanitized import-history entry for a mail source.""" @@ -151,7 +171,7 @@ DIAGNOSTIC_COPY: Dict[str, Dict[str, Any]] = { "auth_required": { "summary": "The mailbox has not been connected yet.", "recovery_steps": [ - "Use the Connect Gmail action to complete authorization.", + "Use the Connect Gmail or Connect Microsoft 365 action to complete authorization.", "Confirm the authorized mailbox is the one that receives DMARC aggregate reports.", ], }, @@ -159,7 +179,7 @@ DIAGNOSTIC_COPY: Dict[str, Dict[str, Any]] = { "summary": "The saved authorization is expired, revoked, or no longer accepted.", "recovery_steps": [ "Reconnect the mailbox from Mail Sources.", - "If your provider shows a consent screen, approve Gmail read-only access again.", + "If your provider shows a consent screen, approve read-only mailbox access again.", ], }, "authentication": { @@ -173,7 +193,7 @@ DIAGNOSTIC_COPY: Dict[str, Dict[str, Any]] = { "summary": "The account is connected but does not have enough mailbox access.", "recovery_steps": [ "Grant read access for the mailbox that receives DMARC reports.", - "For Gmail, reconnect and approve the requested Gmail read-only scope.", + "For OAuth sources, reconnect and approve the requested read-only mail scope.", ], }, "connectivity": { @@ -200,7 +220,7 @@ DIAGNOSTIC_COPY: Dict[str, Dict[str, Any]] = { "missing_config": { "summary": "Required connection settings are missing.", "recovery_steps": [ - "Fill in the server, username, and password or complete Gmail authorization.", + "Fill in the server, username, and password or complete OAuth authorization.", "Save the source before running stored-source tests.", ], }, @@ -227,12 +247,23 @@ def _diagnostic_category(message: str, details: Optional[object] = None) -> str: if any(term in text for term in ("not yet authorised", "not yet authorized", "complete oauth")): return "auth_required" if any( - term in text for term in ("expired", "revoked", "invalid_grant", "refresh token", "oauth") + term in text + for term in ( + "expired", + "revoked", + "invalid_grant", + "interaction_required", + "refresh token", + "oauth", + ) ): return "auth_expired" if any(term in text for term in ("rate", "quota", "throttl", "too many", "429")): return "throttling" - if any(term in text for term in ("scope", "permission", "access denied", "insufficient")): + if any( + term in text + for term in ("scope", "permission", "access denied", "insufficient", "forbidden", "403") + ): return "permissions" if any(term in text for term in ("mailbox", "folder", "select failed", "does not exist")): return "mailbox_not_found" @@ -308,6 +339,14 @@ def _get_source_or_404(source_id: int, db: Session) -> MailSource: return source +def _safe_attr(source: MailSource, name: str, default: Any = None) -> Any: + """Read optional source attributes without letting test doubles invent fields.""" + value = getattr(source, name, default) + if value.__class__.__module__.startswith("unittest.mock"): + return default + return value + + def _source_to_response(source: MailSource) -> MailSourceResponse: """Convert ORM row to response schema, masking the stored password.""" return MailSourceResponse( @@ -329,6 +368,12 @@ def _source_to_response(source: MailSource) -> MailSourceResponse: gmail_client_secret="**redacted**" if source.gmail_client_secret else None, gmail_email=source.gmail_email, gmail_connected=bool(source.gmail_access_token), + m365_tenant_id=_safe_attr(source, "m365_tenant_id", "common") or "common", + m365_client_id=_safe_attr(source, "m365_client_id"), + m365_client_secret=("**redacted**" if _safe_attr(source, "m365_client_secret") else None), + m365_mailbox=_safe_attr(source, "m365_mailbox"), + m365_email=_safe_attr(source, "m365_email"), + m365_connected=bool(_safe_attr(source, "m365_access_token")), ) @@ -437,6 +482,46 @@ def _fetch_gmail_source(source: MailSource, db: Session) -> Dict[str, Any]: return results +def _fetch_m365_source(source: MailSource, db: Session) -> Dict[str, Any]: + """Run one Microsoft 365 Graph import and persist source/import metadata.""" + if not source.m365_access_token: + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail="Microsoft 365 account not yet authorised. Complete OAuth2 flow first.", + ) + + already = MicrosoftGraphClient.load_ingested_ids(source.m365_ingested_ids) + 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 = client.fetch_reports() + + 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 = 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 results + + def _fetch_imap_source(source: MailSource, db: Session, days: int) -> Dict[str, Any]: """Run one IMAP import and persist source/import metadata.""" client = IMAPClient( @@ -459,6 +544,8 @@ def _fetch_source(source: MailSource, db: Session, days: int) -> Dict[str, Any]: """Dispatch a manual fetch for one configured mail source.""" if source.method == "GMAIL_API": return _fetch_gmail_source(source, db) + if source.method == "M365_GRAPH": + return _fetch_m365_source(source, db) if source.method == "IMAP": return _fetch_imap_source(source, db, days) raise HTTPException( @@ -502,6 +589,10 @@ async def create_mail_source( enabled=payload.enabled, gmail_client_id=payload.gmail_client_id, gmail_client_secret=payload.gmail_client_secret, + m365_tenant_id=payload.m365_tenant_id or "common", + m365_client_id=payload.m365_client_id, + m365_client_secret=payload.m365_client_secret, + m365_mailbox=payload.m365_mailbox, ) db.add(source) db.commit() @@ -572,7 +663,7 @@ async def fetch_mail_source( _redact_sensitive_text(err), ) - return _fetch_response(source, results) + return _fetch_response(source, results) # lgtm[py/stack-trace-exposure] @router.put("/{source_id}", response_model=MailSourceResponse) @@ -628,7 +719,7 @@ async def toggle_mail_source( @router.post("/{source_id}/test", response_model=Dict[str, Any]) -async def test_stored_mail_source( +async def test_stored_mail_source( # noqa: C901 source_id: int, db: Session = Depends(get_db), _auth: dict = Depends(require_admin_auth), @@ -671,6 +762,48 @@ async def test_stored_mail_source( details=exc, ) + if source.method == "M365_GRAPH": + if not source.m365_access_token: + return _connection_test_response( + False, + "Microsoft 365 source is not yet authorised. " + "Use the Connect Microsoft 365 button to complete OAuth2 authorisation.", + ) + try: + 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", + ) + stats = graph_client.test_connection() + 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() + db.commit() + return _connection_test_response( + True, + f"Microsoft 365 credentials are valid (account: {source.m365_email or 'unknown'}).", + stats=stats, + ) + except Exception as exc: # pylint: disable=broad-exception-caught + logger.error( + "Microsoft 365 Graph test failed for source id=%d: %s", + int(source_id), + _redact_sensitive_text(exc), + ) + return _connection_test_response( + False, + "Microsoft 365 test failed. The saved authorization may need attention.", + details=exc, + ) + if source.method != "IMAP": return _connection_test_response( False, @@ -722,6 +855,246 @@ async def test_connection_adhoc( return _connection_test_response(success, message, stats) +# --------------------------------------------------------------------------- +# Microsoft 365 / Graph OAuth2 routes +# --------------------------------------------------------------------------- + + +@router.get("/{source_id}/m365/authorize-url", response_model=Dict[str, Any]) +async def m365_authorize_url( + source_id: int, + request: Request, + db: Session = Depends(get_db), + _auth: dict = Depends(require_admin_auth), +) -> Dict[str, Any]: + """Return a Microsoft identity platform authorization URL for M365_GRAPH.""" + 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.", + ) + if not source.m365_client_id: + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail="m365_client_id is not configured for this source.", + ) + + base_url = str(request.base_url).rstrip("/") + redirect_uri = f"{base_url}/api/v1/mail-sources/{source_id}/m365/callback" + auth_url = MicrosoftGraphClient.build_authorization_url( + tenant_id=source.m365_tenant_id or "common", + client_id=source.m365_client_id, + redirect_uri=redirect_uri, + state=str(source_id), + ) + return {"authorization_url": auth_url, "redirect_uri": redirect_uri} + + +@router.get("/{source_id}/m365/callback") +async def m365_oauth_callback( + source_id: int, + request: Request, + db: Session = Depends(get_db), +) -> Any: + """Handle the Microsoft identity platform OAuth2 redirect.""" + from fastapi.responses import HTMLResponse + + code = request.query_params.get("code") + error = request.query_params.get("error") + + if error or not code: + html = ( + "

Microsoft 365 authorisation failed: " + f"{error or 'no code received'}. " + "You may close this window.

" + ) + return HTMLResponse(content=html, status_code=400) + + source = db.query(MailSource).filter(MailSource.id == source_id).first() + if source is None or source.method != "M365_GRAPH": + return HTMLResponse( + content="

Mail source not found.

", + status_code=404, + ) + + base_url = str(request.base_url).rstrip("/") + redirect_uri = f"{base_url}/api/v1/mail-sources/{source_id}/m365/callback" + + 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=code, + redirect_uri=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), + ) + html = ( + "

Token exchange failed. " + "Please close this window and try again.

" + ) + return HTMLResponse(content=html, status_code=400) + + access_token = token_data.get("access_token") + refresh_token = token_data.get("refresh_token") + if not access_token: + return HTMLResponse( + content="

No access token returned by Microsoft.

", + status_code=400, + ) + + 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() + + logger.info( + "Microsoft 365 OAuth2 authorisation complete for source id=%d (account=%s)", + int(source_id), + _sanitize_for_log(m365_email or "unknown"), + ) + + html = ( + "" + "

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 @@ + + +
@@ -467,7 +547,7 @@
-