From 2d32830270df60118c51e86c64198dfd9dd7a870 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Sun, 29 Mar 2026 19:23:33 +0000 Subject: [PATCH 1/9] Initial plan From 25b3567414f545fcba166d242595e4a8c0d18505 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Sun, 29 Mar 2026 19:34:03 +0000 Subject: [PATCH 2/9] feat: implement Gmail API mail source with OAuth2, ingestion tracking, and UI Agent-Logs-Url: https://github.com/christianlouis/dmarq/sessions/0c63851e-a69a-4a76-8dd8-25618e825b8c Co-authored-by: christianlouis <361235+christianlouis@users.noreply.github.com> --- backend/=0.2.0 | 0 backend/=2.0.0 | 0 .../add_gmail_fields_to_mail_sources.py | 40 ++ .../app/api/api_v1/endpoints/mail_sources.py | 349 +++++++++++++++- backend/app/main.py | 203 ++++++++-- backend/app/models/mail_source.py | 13 +- backend/app/services/gmail_client.py | 377 ++++++++++++++++++ backend/app/templates/mail_sources.html | 174 +++++++- backend/app/tests/test_mail_sources.py | 279 +++++++++++++ backend/requirements.txt | 5 +- 10 files changed, 1377 insertions(+), 63 deletions(-) create mode 100644 backend/=0.2.0 create mode 100644 backend/=2.0.0 create mode 100644 backend/alembic/versions/add_gmail_fields_to_mail_sources.py create mode 100644 backend/app/services/gmail_client.py diff --git a/backend/=0.2.0 b/backend/=0.2.0 new file mode 100644 index 0000000..e69de29 diff --git a/backend/=2.0.0 b/backend/=2.0.0 new file mode 100644 index 0000000..e69de29 diff --git a/backend/alembic/versions/add_gmail_fields_to_mail_sources.py b/backend/alembic/versions/add_gmail_fields_to_mail_sources.py new file mode 100644 index 0000000..6e71299 --- /dev/null +++ b/backend/alembic/versions/add_gmail_fields_to_mail_sources.py @@ -0,0 +1,40 @@ +"""add Gmail API fields to mail_sources + +Revision ID: b2c3d4e5f6a7 +Revises: a1b2c3d4e5f6 +Create Date: 2026-03-29 19:00:00.000000 + +""" +from typing import Sequence, Union + +import sqlalchemy as sa +from alembic import op + +# revision identifiers, used by Alembic. +revision: str = "b2c3d4e5f6a7" +down_revision: Union[str, Sequence[str], None] = "a1b2c3d4e5f6" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + """Add Gmail OAuth2 credential columns to mail_sources.""" + op.add_column("mail_sources", sa.Column("gmail_client_id", sa.String(), nullable=True)) + op.add_column("mail_sources", sa.Column("gmail_client_secret", sa.Text(), nullable=True)) + op.add_column("mail_sources", sa.Column("gmail_access_token", sa.Text(), nullable=True)) + op.add_column("mail_sources", sa.Column("gmail_refresh_token", sa.Text(), nullable=True)) + op.add_column("mail_sources", sa.Column("gmail_email", sa.String(), nullable=True)) + op.add_column( + "mail_sources", + sa.Column("gmail_ingested_ids", sa.Text(), nullable=True, server_default="[]"), + ) + + +def downgrade() -> None: + """Remove Gmail OAuth2 credential columns from mail_sources.""" + op.drop_column("mail_sources", "gmail_ingested_ids") + op.drop_column("mail_sources", "gmail_email") + op.drop_column("mail_sources", "gmail_refresh_token") + op.drop_column("mail_sources", "gmail_access_token") + op.drop_column("mail_sources", "gmail_client_secret") + op.drop_column("mail_sources", "gmail_client_id") diff --git a/backend/app/api/api_v1/endpoints/mail_sources.py b/backend/app/api/api_v1/endpoints/mail_sources.py index 5b7ddf8..e79beab 100644 --- a/backend/app/api/api_v1/endpoints/mail_sources.py +++ b/backend/app/api/api_v1/endpoints/mail_sources.py @@ -3,20 +3,22 @@ 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. +persisting anything. Gmail API sources additionally have OAuth2 helper +endpoints (authorize-url, callback, fetch). """ import logging from datetime import datetime from typing import Any, Dict, List, Optional -from fastapi import APIRouter, Depends, HTTPException, status +from fastapi import APIRouter, Depends, HTTPException, Request, status from pydantic import BaseModel from sqlalchemy.orm import Session from app.core.database import get_db from app.core.security import require_admin_auth from app.models.mail_source import MailSource +from app.services.gmail_client import GmailClient from app.services.imap_client import IMAPClient router = APIRouter() @@ -41,6 +43,9 @@ class MailSourceBase(BaseModel): folder: str = "INBOX" polling_interval: int = 60 enabled: bool = True + # Gmail API OAuth2 fields (only relevant when method == GMAIL_API) + gmail_client_id: Optional[str] = None + gmail_client_secret: Optional[str] = None class MailSourceCreate(MailSourceBase): @@ -60,6 +65,8 @@ class MailSourceUpdate(BaseModel): folder: Optional[str] = None polling_interval: Optional[int] = None enabled: Optional[bool] = None + gmail_client_id: Optional[str] = None + gmail_client_secret: Optional[str] = None class MailSourceResponse(MailSourceBase): @@ -71,6 +78,10 @@ class MailSourceResponse(MailSourceBase): updated_at: Optional[datetime] = None # Mask the stored password in responses password: Optional[str] = None + # Gmail: show the authorised email address but not tokens + gmail_email: Optional[str] = None + # Indicate whether OAuth tokens are present (without exposing them) + gmail_connected: bool = False class Config: from_attributes = True @@ -87,6 +98,13 @@ class TestConnectionRequest(BaseModel): method: str = "IMAP" +class GmailCallbackRequest(BaseModel): + """Payload for the Gmail OAuth2 callback endpoint.""" + + code: str + redirect_uri: str + + # --------------------------------------------------------------------------- # Helper # --------------------------------------------------------------------------- @@ -124,6 +142,10 @@ def _source_to_response(source: MailSource) -> MailSourceResponse: last_checked=source.last_checked, created_at=source.created_at, updated_at=source.updated_at, + gmail_client_id=source.gmail_client_id, + gmail_client_secret="**redacted**" if source.gmail_client_secret else None, + gmail_email=source.gmail_email, + gmail_connected=bool(source.gmail_access_token), ) @@ -160,6 +182,8 @@ async def create_mail_source( folder=payload.folder, polling_interval=payload.polling_interval, enabled=payload.enabled, + gmail_client_id=payload.gmail_client_id, + gmail_client_secret=payload.gmail_client_secret, ) db.add(source) db.commit() @@ -242,6 +266,38 @@ async def test_stored_mail_source( """Test the connection for an already-stored mail source using its saved credentials.""" source = _get_source_or_404(source_id, db) + if source.method == "GMAIL_API": + if not source.gmail_access_token: + return { + "success": False, + "message": "Gmail API source is not yet authorised. " + "Use the 'Connect Gmail' button to complete OAuth2 authorisation.", + "timestamp": datetime.now().isoformat(), + } + try: + gmail_client = GmailClient( + client_id=source.gmail_client_id or "", + client_secret=source.gmail_client_secret or "", + access_token=source.gmail_access_token, + refresh_token=source.gmail_refresh_token or "", + ) + # Attempt to list one message to verify the credentials work + service = gmail_client._build_service() # pylint: disable=protected-access + service.users().getProfile(userId="me").execute() + source.last_checked = datetime.utcnow() + db.commit() + return { + "success": True, + "message": f"Gmail API credentials are valid (account: {source.gmail_email or 'unknown'}).", + "timestamp": datetime.now().isoformat(), + } + except Exception as exc: # pylint: disable=broad-exception-caught + return { + "success": False, + "message": f"Gmail API test failed: {exc}", + "timestamp": datetime.now().isoformat(), + } + if source.method != "IMAP": return { "success": False, @@ -308,3 +364,292 @@ async def test_connection_adhoc( "available_mailboxes": stats.get("available_mailboxes", []), "timestamp": datetime.now().isoformat(), } + + +# --------------------------------------------------------------------------- +# Gmail API OAuth2 routes +# --------------------------------------------------------------------------- + + +@router.get("/{source_id}/gmail/authorize-url", response_model=Dict[str, Any]) +async def gmail_authorize_url( + source_id: int, + request: Request, + db: Session = Depends(get_db), + _auth: dict = Depends(require_admin_auth), +) -> Dict[str, Any]: + """ + Return a Google OAuth2 authorization URL for the given GMAIL_API source. + + The frontend should redirect the user to this URL. After the user + grants access Google redirects back to + ``/mail-sources//gmail/callback`` with a ``code`` parameter. + """ + source = _get_source_or_404(source_id, db) + + if source.method != "GMAIL_API": + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail="This endpoint is only available for GMAIL_API sources.", + ) + if not source.gmail_client_id: + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail="gmail_client_id is not configured for this source.", + ) + + # Build a redirect_uri that points back to this server's callback endpoint + base_url = str(request.base_url).rstrip("/") + redirect_uri = f"{base_url}/api/v1/mail-sources/{source_id}/gmail/callback" + + auth_url = GmailClient.build_authorization_url( + client_id=source.gmail_client_id, + redirect_uri=redirect_uri, + state=str(source_id), + ) + return { + "authorization_url": auth_url, + "redirect_uri": redirect_uri, + } + + +@router.get("/{source_id}/gmail/callback") +async def gmail_oauth_callback( + source_id: int, + request: Request, + db: Session = Depends(get_db), +) -> Any: + """ + Handle the Google OAuth2 redirect after the user grants Gmail access. + + Exchanges the authorization ``code`` query parameter for access/refresh + tokens and stores them on the MailSource row. This endpoint is called + directly by Google's redirect, so it does not require the usual API key + authentication; it is protected instead by the state/code being + single-use and bound to the source_id in the URL. + """ + from fastapi.responses import HTMLResponse + + code = request.query_params.get("code") + error = request.query_params.get("error") + + if error or not code: + html = ( + "

Gmail 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 != "GMAIL_API": + 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}/gmail/callback" + + try: + token_data = GmailClient.exchange_code_for_tokens( + client_id=source.gmail_client_id or "", + client_secret=source.gmail_client_secret or "", + code=code, + redirect_uri=redirect_uri, + ) + except ValueError as exc: + logger.error("Gmail token exchange error for source id=%d: %s", source_id, 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 Google.

", + status_code=400, + ) + + gmail_email = GmailClient.get_gmail_email(access_token) + + source.gmail_access_token = access_token + if refresh_token: + source.gmail_refresh_token = refresh_token + if gmail_email: + source.gmail_email = gmail_email + source.updated_at = datetime.utcnow() + db.commit() + + logger.info( + "Gmail OAuth2 authorisation complete for source id=%d (account=%s)", + source_id, + gmail_email or "unknown", + ) + + html = ( + "" + "

✅ Gmail account connected successfully" + f"{(' (' + gmail_email + ')') if gmail_email else ''}. " + "You may close this window.

" + "" + "" + ) + return HTMLResponse(content=html) + + +@router.post("/{source_id}/gmail/callback", response_model=MailSourceResponse) +async def gmail_oauth_callback_post( + source_id: int, + payload: GmailCallbackRequest, + db: Session = Depends(get_db), + _auth: dict = Depends(require_admin_auth), +) -> MailSourceResponse: + """ + Exchange an OAuth2 authorization code for tokens (JSON / programmatic flow). + + This POST variant is for clients that handle the OAuth2 redirect + themselves and post the code here as JSON. Requires the standard + admin authentication. + """ + source = _get_source_or_404(source_id, db) + + if source.method != "GMAIL_API": + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail="This endpoint is only available for GMAIL_API sources.", + ) + + try: + token_data = GmailClient.exchange_code_for_tokens( + client_id=source.gmail_client_id or "", + client_secret=source.gmail_client_secret or "", + code=payload.code, + redirect_uri=payload.redirect_uri, + ) + except ValueError as exc: + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail=str(exc), + ) 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="Google did not return an access token.", + ) + + gmail_email = GmailClient.get_gmail_email(access_token) + + source.gmail_access_token = access_token + if refresh_token: + source.gmail_refresh_token = refresh_token + if gmail_email: + source.gmail_email = gmail_email + source.updated_at = datetime.utcnow() + db.commit() + db.refresh(source) + + logger.info( + "Gmail OAuth2 tokens saved for source id=%d (account=%s)", + source_id, + gmail_email or "unknown", + ) + return _source_to_response(source) + + +@router.post("/{source_id}/gmail/fetch", response_model=Dict[str, Any]) +async def gmail_fetch_reports( + source_id: int, + db: Session = Depends(get_db), + _auth: dict = Depends(require_admin_auth), +) -> Dict[str, Any]: + """ + Manually trigger a Gmail DMARC report fetch for the given source. + + Searches Gmail for emails matching the DMARC report heuristic, ingests + any attachments not yet seen, and returns a summary. + """ + source = _get_source_or_404(source_id, db) + + if source.method != "GMAIL_API": + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail="This endpoint is only available for GMAIL_API sources.", + ) + if not source.gmail_access_token: + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail="Gmail account not yet authorised. Complete OAuth2 flow first.", + ) + + already = GmailClient.load_ingested_ids(source.gmail_ingested_ids) + client = GmailClient( + client_id=source.gmail_client_id or "", + client_secret=source.gmail_client_secret or "", + access_token=source.gmail_access_token, + refresh_token=source.gmail_refresh_token or "", + already_ingested_ids=already, + ) + + results = client.fetch_reports() + + # Persist updated ingested IDs and any refreshed tokens + if results.get("new_ingested_ids"): + all_ids = list(dict.fromkeys(already + results["new_ingested_ids"])) + source.gmail_ingested_ids = GmailClient.dump_ingested_ids(all_ids) + + refreshed = client.get_refreshed_tokens() + if refreshed: + source.gmail_access_token = refreshed["access_token"] + if "refresh_token" in refreshed: + source.gmail_refresh_token = refreshed["refresh_token"] + + source.last_checked = datetime.utcnow() + db.commit() + + logger.info( + "Gmail fetch for source id=%d: processed=%d reports_found=%d", + source_id, + results.get("processed", 0), + results.get("reports_found", 0), + ) + + return { + "success": results.get("success", False), + "processed": results.get("processed", 0), + "reports_found": results.get("reports_found", 0), + "new_domains": results.get("new_domains", []), + "errors": results.get("errors", []) or None, + "timestamp": datetime.now().isoformat(), + } + + +@router.delete("/{source_id}/gmail/connection", status_code=status.HTTP_204_NO_CONTENT) +async def gmail_disconnect( + source_id: int, + db: Session = Depends(get_db), + _auth: dict = Depends(require_admin_auth), +) -> None: + """Revoke / clear the stored Gmail OAuth2 tokens for this source.""" + source = _get_source_or_404(source_id, db) + + if source.method != "GMAIL_API": + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail="This endpoint is only available for GMAIL_API sources.", + ) + + source.gmail_access_token = None + source.gmail_refresh_token = None + source.gmail_email = None + source.updated_at = datetime.utcnow() + db.commit() + logger.info("Gmail tokens cleared for source id=%d", source_id) diff --git a/backend/app/main.py b/backend/app/main.py index e27b1fc..4093afc 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -15,6 +15,7 @@ from app.core.database import Base, SessionLocal, engine from app.core.security import add_api_key, generate_api_key, require_admin_auth from app.middleware.security import SecurityHeadersMiddleware from app.models.mail_source import MailSource # noqa: F401 – ensure table is registered +from app.services.gmail_client import GmailClient from app.services.imap_client import IMAPClient from app.services.report_store import ReportStore @@ -69,6 +70,66 @@ def _poll_single_imap_source(source: MailSource) -> None: ) +def _poll_single_gmail_source(source: MailSource) -> None: + """Fetch DMARC reports for a single GMAIL_API mail source.""" + global last_check_time # pylint: disable=global-statement + + if not source.gmail_access_token: + logger.info( + "Gmail polling (source id=%d): skipped – OAuth2 not yet authorised", + source.id, + ) + return + + already = GmailClient.load_ingested_ids(source.gmail_ingested_ids) + client = GmailClient( + client_id=source.gmail_client_id or "", + client_secret=source.gmail_client_secret or "", + access_token=source.gmail_access_token, + refresh_token=source.gmail_refresh_token or "", + already_ingested_ids=already, + ) + + results = client.fetch_reports() + + db = SessionLocal() + try: + src = db.query(MailSource).get(source.id) + if src: + if results.get("new_ingested_ids"): + all_ids = list(dict.fromkeys(already + results["new_ingested_ids"])) + src.gmail_ingested_ids = GmailClient.dump_ingested_ids(all_ids) + + refreshed = client.get_refreshed_tokens() + if refreshed: + src.gmail_access_token = refreshed["access_token"] + if "refresh_token" in refreshed: + src.gmail_refresh_token = refreshed["refresh_token"] + + src.last_checked = datetime.utcnow() + db.commit() + finally: + db.close() + + last_check_time = datetime.now() + + if results["success"]: + logger.info( + "Gmail polling (source id=%d): %s emails processed, %s 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( + "Gmail polling (source id=%d) failed: %s", + source.id, + results.get("error", "Unknown error"), + ) + + def _poll_all_enabled_sources() -> None: """Iterate over all enabled mail sources and poll each one.""" db = SessionLocal() @@ -84,17 +145,22 @@ def _poll_all_enabled_sources() -> None: return for source in enabled_sources: - if source.method != "IMAP": + if source.method == "GMAIL_API": + try: + _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 == "IMAP": + try: + _poll_single_imap_source(source) + except Exception as e: # pylint: disable=broad-exception-caught + logger.error("Error polling mail source id=%d: %s", source.id, str(e)) + else: logger.info( "Skipping mail source id=%d method=%r (not yet implemented)", source.id, source.method, ) - continue - try: - _poll_single_imap_source(source) - except Exception as e: # pylint: disable=broad-exception-caught - logger.error("Error polling mail source id=%d: %s", source.id, str(e)) def _next_sleep_seconds(min_sleep: int = 60) -> int: @@ -386,7 +452,95 @@ async def trigger_imap_poll(auth: dict = Depends(require_admin_auth)): } for source in enabled_sources: - if source.method != "IMAP": + if source.method == "GMAIL_API": + if not source.gmail_access_token: + results_summary.append( + { + "source_id": source.id, + "name": source.name, + "skipped": True, + "reason": "Gmail account not yet authorised", + } + ) + continue + try: + already = GmailClient.load_ingested_ids(source.gmail_ingested_ids) + gmail_client = GmailClient( + client_id=source.gmail_client_id or "", + client_secret=source.gmail_client_secret or "", + access_token=source.gmail_access_token, + refresh_token=source.gmail_refresh_token or "", + already_ingested_ids=already, + ) + results = gmail_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.gmail_ingested_ids = GmailClient.dump_ingested_ids(all_ids) + refreshed = gmail_client.get_refreshed_tokens() + if refreshed: + source.gmail_access_token = refreshed["access_token"] + if "refresh_token" in refreshed: + source.gmail_refresh_token = refreshed["refresh_token"] + source.last_checked = datetime.utcnow() + db.commit() + + results_summary.append( + { + "source_id": source.id, + "name": source.name, + "success": results["success"], + "processed": results.get("processed", 0), + "reports_found": results.get("reports_found", 0), + "new_domains": results.get("new_domains", []), + } + ) + except Exception as e: # pylint: disable=broad-exception-caught + logger.error("Error polling Gmail source id=%d: %s", source.id, str(e)) + results_summary.append( + { + "source_id": source.id, + "name": source.name, + "success": False, + "error": "Failed to poll. Check server logs for details.", + } + ) + elif source.method == "IMAP": + try: + imap_client = IMAPClient( + server=source.server, + port=source.port or 993, + username=source.username, + password=source.password, + delete_emails=False, + ) + results = imap_client.fetch_reports(days=7) + last_check_time = datetime.now() + source.last_checked = datetime.utcnow() + db.commit() + + results_summary.append( + { + "source_id": source.id, + "name": source.name, + "success": results["success"], + "processed": results.get("processed", 0), + "reports_found": results.get("reports_found", 0), + "new_domains": results.get("new_domains", []), + } + ) + except Exception as e: # pylint: disable=broad-exception-caught + logger.error("Error polling mail source id=%d: %s", source.id, str(e)) + results_summary.append( + { + "source_id": source.id, + "name": source.name, + "success": False, + "error": "Failed to poll. Check server logs for details.", + } + ) + else: results_summary.append( { "source_id": source.id, @@ -395,41 +549,6 @@ async def trigger_imap_poll(auth: dict = Depends(require_admin_auth)): "reason": f"method '{source.method}' not yet implemented", } ) - continue - - try: - imap_client = IMAPClient( - server=source.server, - port=source.port or 993, - username=source.username, - password=source.password, - delete_emails=False, - ) - results = imap_client.fetch_reports(days=7) - last_check_time = datetime.now() - source.last_checked = datetime.utcnow() - db.commit() - - results_summary.append( - { - "source_id": source.id, - "name": source.name, - "success": results["success"], - "processed": results.get("processed", 0), - "reports_found": results.get("reports_found", 0), - "new_domains": results.get("new_domains", []), - } - ) - except Exception as e: # pylint: disable=broad-exception-caught - logger.error("Error polling mail source id=%d: %s", source.id, str(e)) - results_summary.append( - { - "source_id": source.id, - "name": source.name, - "success": False, - "error": "Failed to poll. Check server logs for details.", - } - ) finally: db.close() diff --git a/backend/app/models/mail_source.py b/backend/app/models/mail_source.py index 75fde33..7bd2517 100644 --- a/backend/app/models/mail_source.py +++ b/backend/app/models/mail_source.py @@ -15,7 +15,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 (stub for future implementation) + - ``GMAIL_API`` – Gmail API with OAuth 2.0 """ __tablename__ = "mail_sources" @@ -38,6 +38,17 @@ class MailSource(Base): use_ssl = Column(Boolean, default=True) folder = Column(String, default="INBOX") + # Gmail API OAuth2 credentials (used by GMAIL_API method) + # NOTE: tokens stored in plaintext – encrypt at the app layer in production. + gmail_client_id = Column(String, nullable=True) + gmail_client_secret = Column(Text, nullable=True) + gmail_access_token = Column(Text, nullable=True) + gmail_refresh_token = Column(Text, nullable=True) + # Email address of the authorised Gmail account + gmail_email = Column(String, nullable=True) + # JSON-encoded list of Gmail message IDs that have already been ingested + gmail_ingested_ids = Column(Text, nullable=True, default="[]") + # Polling behaviour polling_interval = Column(Integer, default=60) # minutes diff --git a/backend/app/services/gmail_client.py b/backend/app/services/gmail_client.py new file mode 100644 index 0000000..88282f1 --- /dev/null +++ b/backend/app/services/gmail_client.py @@ -0,0 +1,377 @@ +""" +Gmail API client for retrieving DMARC reports. + +Connects to Gmail via OAuth 2.0, searches for emails that are likely to +contain DMARC aggregate-report attachments, and processes any new ones. +Already-ingested message IDs are tracked so the same email is never +processed twice (no messages are modified or deleted). +""" + +import base64 +import email +import json +import logging +from typing import Any, Dict, List, Optional, Tuple +from urllib.parse import urlencode + +import httpx +from google.auth.transport.requests import Request +from google.oauth2.credentials import Credentials +from googleapiclient.discovery import build +from googleapiclient.errors import HttpError + +from app.services.dmarc_parser import DMARCParser +from app.services.report_store import ReportStore + +logger = logging.getLogger(__name__) + +# --------------------------------------------------------------------------- +# OAuth2 scopes – read-only access to Gmail messages is all we need +# --------------------------------------------------------------------------- + +GMAIL_SCOPES = [ + "https://www.googleapis.com/auth/gmail.readonly", +] + +# --------------------------------------------------------------------------- +# Gmail search query used to find emails likely containing DMARC reports. +# +# Strategy: +# • Require at least one attachment whose name ends in .zip, .gz, or .xml +# (the three formats used by virtually every DMARC sender). +# • Additionally require *either* a keyword in the subject that DMARC senders +# use, or an envelope-from that belongs to a well-known DMARC reporting +# address. This keeps false-positive rates low while catching reports +# from providers that don't follow naming conventions perfectly. +# --------------------------------------------------------------------------- + +DMARC_GMAIL_QUERY = ( + "has:attachment " + "(filename:zip OR filename:gz OR filename:xml) " + "(subject:dmarc OR subject:report OR subject:rua " + 'OR subject:"aggregate report" OR subject:"domain report" ' + "OR from:dmarc OR from:dmarc-noreply OR from:reports OR from:postmaster)" +) + +# How many message results to fetch per API page +_PAGE_SIZE = 100 + + +class GmailClient: + """ + Client for retrieving DMARC reports from a Gmail account via the Gmail API. + + OAuth2 tokens are accepted at construction time and auto-refreshed when + expired. The caller is responsible for persisting any refreshed tokens + returned by :meth:`get_refreshed_tokens`. + """ + + def __init__( + self, + client_id: str, + client_secret: str, + access_token: str, + refresh_token: str, + already_ingested_ids: Optional[List[str]] = None, + ): + self.client_id = client_id + self.client_secret = client_secret + self._initial_access_token = access_token + self.already_ingested_ids: List[str] = list(already_ingested_ids or []) + self.report_store = ReportStore.get_instance() + + self.credentials = Credentials( + token=access_token, + refresh_token=refresh_token, + token_uri="https://oauth2.googleapis.com/token", + client_id=client_id, + client_secret=client_secret, + scopes=GMAIL_SCOPES, + ) + + # ------------------------------------------------------------------ + # Public helpers + # ------------------------------------------------------------------ + + def get_refreshed_tokens(self) -> Optional[Dict[str, str]]: + """ + Return updated tokens if the google-auth library has refreshed them. + + Call this after :meth:`fetch_reports` and persist any non-None result + so the next run doesn't need an extra refresh round-trip. + """ + current = self.credentials.token + if current and current != self._initial_access_token: + result: Dict[str, str] = {"access_token": current} + if self.credentials.refresh_token: + result["refresh_token"] = self.credentials.refresh_token + return result + return None + + # ------------------------------------------------------------------ + # OAuth2 helpers (static / class methods used by the endpoint layer) + # ------------------------------------------------------------------ + + @staticmethod + def build_authorization_url( + client_id: str, + redirect_uri: str, + state: Optional[str] = None, + ) -> str: + """ + Construct the Google OAuth2 authorization URL. + + Requests offline access so a refresh token is issued, and forces + the consent screen so the refresh token is always returned even if + the user has authorised this app before. + """ + params: Dict[str, str] = { + "client_id": client_id, + "response_type": "code", + "scope": " ".join(GMAIL_SCOPES), + "redirect_uri": redirect_uri, + "access_type": "offline", + "prompt": "consent", + } + if state: + params["state"] = state + return "https://accounts.google.com/o/oauth2/v2/auth?" + urlencode(params) + + @staticmethod + def exchange_code_for_tokens( + client_id: str, + client_secret: str, + code: str, + redirect_uri: str, + ) -> Dict[str, Any]: + """ + Synchronously exchange an authorization code for access+refresh tokens. + + Returns the raw JSON from Google's token endpoint. The caller + should check for ``access_token`` in the result before using it. + + Raises: + ValueError: if Google returns a non-200 response. + """ + resp = httpx.post( + "https://oauth2.googleapis.com/token", + data={ + "code": code, + "client_id": client_id, + "client_secret": client_secret, + "redirect_uri": redirect_uri, + "grant_type": "authorization_code", + }, + ) + if resp.status_code != 200: + raise ValueError(f"Token exchange failed ({resp.status_code}): {resp.text}") + return resp.json() + + @staticmethod + def get_gmail_email(access_token: str) -> Optional[str]: + """ + Return the email address associated with an access token. + + Uses the OAuth2 userinfo endpoint. Returns None on failure. + """ + try: + resp = httpx.get( + "https://www.googleapis.com/oauth2/v2/userinfo", + headers={"Authorization": f"Bearer {access_token}"}, + ) + if resp.status_code == 200: + return resp.json().get("email") + except Exception as exc: # pylint: disable=broad-exception-caught + logger.error("Failed to fetch Gmail email address: %s", exc) + return None + + # ------------------------------------------------------------------ + # Core fetching logic + # ------------------------------------------------------------------ + + def fetch_reports(self) -> Dict[str, Any]: + """ + Search Gmail for DMARC report emails and ingest any new ones. + + Emails that have already been ingested (tracked via + ``already_ingested_ids``) are silently skipped. No messages are + modified or deleted. + + Returns: + A dict with keys ``success``, ``processed``, ``reports_found``, + ``new_domains``, ``errors``, and ``new_ingested_ids`` (the IDs + added in this run so the caller can persist them). + """ + stats: Dict[str, Any] = { + "success": True, + "processed": 0, + "reports_found": 0, + "new_domains": [], + "errors": [], + "new_ingested_ids": [], + } + + try: + service = self._build_service() + except Exception as exc: # pylint: disable=broad-exception-caught + logger.error("Gmail API: failed to build service: %s", exc) + return {**stats, "success": False, "error": str(exc)} + + try: + message_ids = self._list_dmarc_message_ids(service) + except Exception as exc: # pylint: disable=broad-exception-caught + logger.error("Gmail API: failed to list messages: %s", exc) + return {**stats, "success": False, "error": str(exc)} + + domains_before = set(self.report_store.get_domains()) + + for msg_id in message_ids: + if msg_id in self.already_ingested_ids: + continue + + stats["processed"] += 1 + found = self._process_message(service, msg_id, stats) + if found > 0: + stats["new_ingested_ids"].append(msg_id) + self.already_ingested_ids.append(msg_id) + else: + # Track it anyway so we don't re-examine it next run + stats["new_ingested_ids"].append(msg_id) + self.already_ingested_ids.append(msg_id) + + domains_after = set(self.report_store.get_domains()) + stats["new_domains"] = list(domains_after - domains_before) + return stats + + # ------------------------------------------------------------------ + # Private helpers + # ------------------------------------------------------------------ + + def _build_service(self): + """Build (and auto-refresh if needed) the Gmail API service object.""" + if self.credentials.expired and self.credentials.refresh_token: + try: + self.credentials.refresh(Request()) + except Exception as exc: # pylint: disable=broad-exception-caught + logger.error("Gmail token refresh failed: %s", exc) + raise + + return build("gmail", "v1", credentials=self.credentials, cache_discovery=False) + + def _list_dmarc_message_ids(self, service) -> List[str]: + """Return all Gmail message IDs matching the DMARC search query.""" + ids: List[str] = [] + page_token: Optional[str] = None + + while True: + kwargs: Dict[str, Any] = { + "userId": "me", + "q": DMARC_GMAIL_QUERY, + "maxResults": _PAGE_SIZE, + } + if page_token: + kwargs["pageToken"] = page_token + + try: + result = service.users().messages().list(**kwargs).execute() + except HttpError as exc: + logger.error("Gmail API list error: %s", exc) + raise + + for msg in result.get("messages", []): + ids.append(msg["id"]) + + page_token = result.get("nextPageToken") + if not page_token: + break + + return ids + + def _process_message(self, service, msg_id: str, stats: dict) -> int: + """ + Download a Gmail message and process any DMARC-report attachments. + + Returns the number of DMARC reports found in this message. + """ + try: + msg_data = ( + service.users() + .messages() + .get(userId="me", id=msg_id, format="raw") + .execute() + ) + except HttpError as exc: + logger.error("Gmail API: failed to fetch message %s: %s", msg_id, exc) + stats["errors"].append(f"Failed to fetch message {msg_id}") + return 0 + + raw_bytes = base64.urlsafe_b64decode(msg_data.get("raw", "")) + msg = email.message_from_bytes(raw_bytes) + return self._process_attachments(msg, stats) + + def _process_attachments(self, msg: email.message.Message, stats: dict) -> int: + """Walk a parsed email message and extract DMARC report attachments.""" + reports_found = 0 + + for part in msg.walk(): + if part.get_content_disposition() != "attachment": + continue + + filename = part.get_filename() or "" + if hasattr(filename, "encode"): + # Decode RFC 2047-encoded filenames + from email.header import decode_header + + parts = decode_header(filename) + decoded_parts = [] + for raw, charset in parts: + if isinstance(raw, bytes): + decoded_parts.append(raw.decode(charset or "utf-8", errors="replace")) + else: + decoded_parts.append(raw) + filename = "".join(decoded_parts) + + lower = filename.lower() + if not ( + lower.endswith(".xml") + or lower.endswith(".zip") + or lower.endswith(".gz") + or lower.endswith(".gzip") + ): + continue + + content = part.get_payload(decode=True) + if not content: + continue + + try: + parser = DMARCParser() + reports = parser.parse(content, filename) + for report in reports: + self.report_store.add_report(report) + stats["reports_found"] += 1 + reports_found += 1 + except Exception as exc: # pylint: disable=broad-exception-caught + logger.error("Failed to parse DMARC attachment %s: %s", filename, exc) + stats["errors"].append(f"Failed to parse {filename}: {exc}") + + return reports_found + + # ------------------------------------------------------------------ + # Convenience: load / save ingested IDs from/to the JSON text column + # ------------------------------------------------------------------ + + @staticmethod + def load_ingested_ids(json_text: Optional[str]) -> List[str]: + """Deserialise the gmail_ingested_ids text column into a list.""" + if not json_text: + return [] + try: + return json.loads(json_text) + except (json.JSONDecodeError, TypeError): + return [] + + @staticmethod + def dump_ingested_ids(ids: List[str]) -> str: + """Serialise the list of ingested IDs back to a JSON string.""" + return json.dumps(ids) diff --git a/backend/app/templates/mail_sources.html b/backend/app/templates/mail_sources.html index 40da181..b7a1e9d 100644 --- a/backend/app/templates/mail_sources.html +++ b/backend/app/templates/mail_sources.html @@ -61,8 +61,8 @@ Name Method - Server - Username + Server / Account + User / Status Last Checked Enabled Actions @@ -75,8 +75,8 @@ - - + + - + @@ -229,6 +229,72 @@ + + +
@@ -252,18 +318,34 @@
- + +
+ + +