feat: define mail connector framework

backend/app/core/redaction.py

@@ -7,11 +7,14 @@ import re
 SENSITIVE_VALUE = "**redacted**"
 
 _SENSITIVE_KEY_PATTERN = re.compile(
-    r"(?i)([\"']?\b(?:access_token|api_key|apikey|authorization|bearer|client_secret|"
+    r"(?i)([\"']?\b(?:access_token|api_key|apikey|bearer|client_secret|"
     r"gmail_client_secret|id_token|passwd|password|refresh_token|secret|token)\b[\"']?"
     r"\s*[:=]\s*[\"']?)([^\"'\s,;&}]+)([\"']?)"
 )
 _BEARER_TOKEN_PATTERN = re.compile(r"(?i)\b(bearer)\s+([A-Za-z0-9._~+/=-]{8,})")
+_AUTHORIZATION_BEARER_PATTERN = re.compile(
+    r"(?i)\b(authorization\s*[:=]\s*bearer)\s+([A-Za-z0-9._~+/=-]{8,})"
+)
 
 
 def sanitize_for_log(value: object) -> str:
@@ -22,5 +25,6 @@ def sanitize_for_log(value: object) -> str:
 def redact_sensitive_text(value: object) -> str:
     """Sanitize text and redact common secret-bearing key/value fragments."""
     text = sanitize_for_log(value)
+    text = _AUTHORIZATION_BEARER_PATTERN.sub(r"\1 " + SENSITIVE_VALUE, text)
     text = _SENSITIVE_KEY_PATTERN.sub(r"\1" + SENSITIVE_VALUE + r"\3", text)
     return _BEARER_TOKEN_PATTERN.sub(r"\1 " + SENSITIVE_VALUE, text)

backend/app/services/gmail_client.py

@@ -9,7 +9,6 @@ processed twice (no messages are modified or deleted).
 
 import base64
 import email
-import json
 import logging
 from typing import Any, Dict, List, Optional
 from urllib.parse import urlencode
@@ -24,6 +23,14 @@ from app.services.dmarc_parser import DMARCParser
 from app.services.forensic_parser import ForensicParser
 from app.services.forensic_persistence import forensic_report_exists, save_forensic_report
 from app.services.forensic_redaction import get_forensic_redaction_policy
+from app.services.mail_connector import (
+    append_import_detail,
+    connector_failure_stats,
+    dump_ingested_ids,
+    initial_import_stats,
+    load_ingested_ids,
+    sanitize_connector_error,
+)
 from app.services.report_persistence import report_exists, save_parsed_report
 from app.services.report_store import ReportStore
 
@@ -210,30 +217,19 @@ class GmailClient:
             ``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,
-            "forensic_reports_found": 0,
-            "duplicate_reports": 0,
-            "duplicate_forensic_reports": 0,
-            "new_domains": [],
-            "errors": [],
-            "new_ingested_ids": [],
-            "details": [],
-        }
+        stats = initial_import_stats()
 
         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)}
+            return connector_failure_stats(stats, "Failed to initialize Gmail API.", error=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)}
+            return connector_failure_stats(stats, "Failed to list Gmail messages.", error=exc)
 
         domains_before = set(self.report_store.get_domains())
 
@@ -306,9 +302,7 @@ class GmailClient:
     @staticmethod
     def _append_detail(stats: dict, **detail: str) -> None:
         """Append a compact attachment/message outcome to the import stats."""
-        stats.setdefault("details", []).append(
-            {key: value for key, value in detail.items() if value}
-        )
+        append_import_detail(stats, **detail)
 
     def _process_message(self, service, msg_id: str, stats: dict) -> int:
         """
@@ -322,7 +316,7 @@ class GmailClient:
             )
         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}")
+            stats["errors"].append(sanitize_connector_error(f"Failed to fetch message {msg_id}"))
             self._append_detail(
                 stats,
                 status="error",
@@ -443,7 +437,7 @@ class GmailClient:
         except Exception as exc:  # pylint: disable=broad-exception-caught
             logger.error("Failed to parse Gmail forensic report %s: %s", message_id, exc)
             stats.setdefault("errors", []).append(
-                f"Failed to parse forensic report {message_id}: {exc}"
+                sanitize_connector_error(f"Failed to parse forensic report {message_id}: {exc}")
             )
             self._append_detail(
                 stats,
@@ -520,7 +514,9 @@ class GmailClient:
                     )
             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}")
+                stats["errors"].append(
+                    sanitize_connector_error(f"Failed to parse {filename}: {exc}")
+                )
                 self._append_detail(
                     stats,
                     status="error",
@@ -539,14 +535,9 @@ class GmailClient:
     @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 []
+        return load_ingested_ids(json_text)
 
     @staticmethod
     def dump_ingested_ids(ids: List[str]) -> str:
         """Serialise the list of ingested IDs back to a JSON string."""
-        return json.dumps(ids)
+        return dump_ingested_ids(ids)

backend/app/services/imap_client.py

@@ -10,6 +10,11 @@ from app.services.dmarc_parser import DMARCParser
 from app.services.forensic_parser import ForensicParser
 from app.services.forensic_persistence import forensic_report_exists, save_forensic_report
 from app.services.forensic_redaction import get_forensic_redaction_policy
+from app.services.mail_connector import (
+    append_import_detail,
+    initial_import_stats,
+    sanitize_connector_error,
+)
 from app.services.report_persistence import report_exists, save_parsed_report
 from app.services.report_store import ReportStore
 
@@ -255,18 +260,7 @@ class IMAPClient:
             logger.error("IMAP credentials not fully configured")
             return {"success": False, "error": "IMAP credentials not configured", "processed": 0}
 
-        stats = {
-            "success": True,
-            "processed": 0,
-            "reports_found": 0,
-            "forensic_reports_found": 0,
-            "deleted": 0,
-            "duplicate_reports": 0,
-            "duplicate_forensic_reports": 0,
-            "new_domains": [],
-            "errors": [],
-            "details": [],
-        }
+        stats = initial_import_stats(deleted=True)
 
         try:
             # Connect to the mail server
@@ -317,6 +311,7 @@ class IMAPClient:
                 "success": False,
                 "error": "Error connecting to mailbox. Check server logs for details.",
                 "processed": 0,
+                "errors": [sanitize_connector_error(e)],
             }
 
     def _is_dmarc_report_email(self, msg: email.message.Message) -> bool:
@@ -446,9 +441,7 @@ class IMAPClient:
         """Append a compact attachment/message outcome to the import stats."""
         if stats is None:
             return
-        stats.setdefault("details", []).append(
-            {key: value for key, value in detail.items() if value}
-        )
+        append_import_detail(stats, **detail)
 
     def _store_report_if_new(
         self,
@@ -560,7 +553,7 @@ class IMAPClient:
             logger.error("Error processing forensic report email %s: %s", message_id, exc)
             if stats is not None:
                 stats.setdefault("errors", []).append(
-                    f"Failed to parse forensic report {message_id}: {exc}"
+                    sanitize_connector_error(f"Failed to parse forensic report {message_id}: {exc}")
                 )
             self._append_detail(
                 stats,
@@ -604,7 +597,9 @@ class IMAPClient:
         except Exception as exc:  # pylint: disable=broad-exception-caught
             logger.error("Error processing attachment %s: %s", filename, str(exc))
             if stats is not None:
-                stats.setdefault("errors", []).append(f"Failed to parse {filename}: {exc}")
+                stats.setdefault("errors", []).append(
+                    sanitize_connector_error(f"Failed to parse {filename}: {exc}")
+                )
             self._append_detail(
                 stats,
                 status="error",

backend/app/services/mail_connector.py

@@ -0,0 +1,164 @@
+"""Shared contracts and helpers for mail-source connectors."""
+
+from __future__ import annotations
+
+import json
+from dataclasses import dataclass
+from typing import Any, Dict, Iterable, List, Optional, Protocol
+
+from app.core.redaction import redact_sensitive_text
+
+MAX_CONNECTOR_ERROR_LENGTH = 500
+
+
+@dataclass(frozen=True)
+class ConnectorImportContext:
+    """Safe import context that can be returned to operators and import history."""
+
+    source_type: str
+    mailbox: Optional[str] = None
+    folder: Optional[str] = None
+    search_window_days: Optional[int] = None
+
+    def as_stats(self) -> Dict[str, Any]:
+        stats: Dict[str, Any] = {"source_type": self.source_type}
+        if self.mailbox:
+            stats["target_mailbox"] = self.mailbox
+        if self.folder:
+            stats["target_folder"] = self.folder
+        if self.search_window_days is not None:
+            stats["search_window_days"] = self.search_window_days
+        return stats
+
+
+@dataclass(frozen=True)
+class ConnectorMessage:
+    """Provider-neutral message metadata used by connector implementations."""
+
+    message_id: str
+    subject: str = ""
+    sender: str = ""
+    received_at: Optional[str] = None
+    has_attachments: bool = False
+    raw: Any = None
+
+
+@dataclass(frozen=True)
+class ConnectorAttachment:
+    """Provider-neutral attachment payload used by connector implementations."""
+
+    filename: str
+    content: bytes
+    content_type: str = ""
+    raw: Any = None
+
+
+class MailSourceConnector(Protocol):
+    """Interface new mailbox connectors should satisfy before endpoint wiring."""
+
+    def import_context(self, days: Optional[int] = None) -> ConnectorImportContext:
+        """Return safe, non-secret context for history and API responses."""
+
+    def search_messages(self, days: int) -> Iterable[Any]:
+        """Return provider messages in the requested search window."""
+
+    def iter_attachments(self, message: Any) -> Iterable[Any]:
+        """Yield provider attachments for one message."""
+
+    def fetch_reports(self, days: int = 7) -> Dict[str, Any]:
+        """Fetch, parse, and persist DMARC reports."""
+
+
+def clamp_search_window(days: Optional[int], *, default: int = 7, maximum: int = 365) -> int:
+    """Normalize user-supplied backfill windows for connector fetches."""
+    try:
+        value = int(days or default)
+    except (TypeError, ValueError):
+        value = default
+    return max(1, min(value, maximum))
+
+
+def initial_import_stats(
+    context: Optional[ConnectorImportContext] = None,
+    *,
+    deleted: bool = False,
+) -> Dict[str, Any]:
+    """Return the shared import-result shape used by mailbox connectors."""
+    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": [],
+    }
+    if deleted:
+        stats["deleted"] = 0
+    if context:
+        stats.update(context.as_stats())
+    return stats
+
+
+def append_import_detail(
+    stats: Optional[Dict[str, Any]],
+    *,
+    context: Optional[ConnectorImportContext] = None,
+    **detail: Any,
+) -> None:
+    """Append one compact, sanitized message or attachment outcome."""
+    if stats is None:
+        return
+    if context:
+        detail.setdefault("mailbox", context.mailbox)
+        detail.setdefault("folder", context.folder)
+    clean_detail = {
+        str(key): sanitize_connector_error(value)
+        for key, value in detail.items()
+        if value not in (None, "")
+    }
+    if clean_detail:
+        stats.setdefault("details", []).append(clean_detail)
+
+
+def sanitize_connector_error(value: object) -> str:
+    """Return a compact, log-safe connector diagnostic with secrets redacted."""
+    text = redact_sensitive_text(value).strip()
+    if len(text) > MAX_CONNECTOR_ERROR_LENGTH:
+        return text[: MAX_CONNECTOR_ERROR_LENGTH - 3] + "..."
+    return text
+
+
+def connector_failure_stats(
+    stats: Dict[str, Any],
+    message: str,
+    *,
+    error: Optional[object] = None,
+) -> Dict[str, Any]:
+    """Return a standardized failed import payload for provider errors."""
+    safe_message = sanitize_connector_error(error if error is not None else message)
+    return {
+        **stats,
+        "success": False,
+        "error": safe_message,
+        "errors": [safe_message],
+    }
+
+
+def load_ingested_ids(json_text: Optional[str]) -> List[str]:
+    """Deserialize a connector ingested-message-id JSON column."""
+    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 []
+
+
+def dump_ingested_ids(ids: Iterable[Any]) -> str:
+    """Serialize connector ingested-message IDs for database storage."""
+    return json.dumps([str(item) for item in ids])

backend/app/services/microsoft_graph_client.py

@@ -1,16 +1,26 @@
 """Microsoft Graph client for retrieving DMARC aggregate reports."""
 
 import base64
-import json
 import logging
 import time
 from datetime import datetime, timedelta
-from typing import Any, Callable, Dict, List, Optional
+from typing import Any, Callable, Dict, Iterable, List, Optional
 from urllib.parse import quote, urlencode
 
 import httpx
 
 from app.services.dmarc_parser import DMARCParser
+from app.services.mail_connector import (
+    ConnectorImportContext,
+    MailSourceConnector,
+    append_import_detail,
+    clamp_search_window,
+    connector_failure_stats,
+    dump_ingested_ids,
+    initial_import_stats,
+    load_ingested_ids,
+    sanitize_connector_error,
+)
 from app.services.report_persistence import report_exists, save_parsed_report
 from app.services.report_store import ReportStore
 
@@ -49,7 +59,7 @@ class MicrosoftGraphError(RuntimeError):
     """Raised when Microsoft Graph or the token endpoint returns a failure."""
 
 
-class MicrosoftGraphClient:
+class MicrosoftGraphClient(MailSourceConnector):
     """
     Retrieve DMARC aggregate reports from Microsoft 365 through Microsoft Graph.
 
@@ -155,18 +165,12 @@ class MicrosoftGraphClient:
     @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 []
+        return load_ingested_ids(json_text)
 
     @staticmethod
     def dump_ingested_ids(ids: List[str]) -> str:
         """Serialize Graph message IDs for database storage."""
-        return json.dumps(ids)
+        return dump_ingested_ids(ids)
 
     def test_connection(self) -> Dict[str, Any]:
         """Verify that the saved delegated token can read the target mailbox."""
@@ -234,30 +238,36 @@ class MicrosoftGraphClient:
             url = data.get("@odata.nextLink")
             params = None
 
+    def import_context(self, days: Optional[int] = None) -> ConnectorImportContext:
+        """Return safe Microsoft 365 import context for API responses/history."""
+        return ConnectorImportContext(
+            source_type="M365_GRAPH",
+            mailbox=self._target_mailbox_label(),
+            folder=self._target_folder_label(),
+            search_window_days=days,
+        )
+
+    def search_messages(self, days: int) -> Iterable[Dict[str, Any]]:
+        """Return Microsoft Graph messages that look like DMARC reports."""
+        return self._list_dmarc_messages(days=days)
+
+    def iter_attachments(self, message: Dict[str, Any]) -> Iterable[Dict[str, Any]]:
+        """Yield Microsoft Graph attachments for one message."""
+        message_id = str(message.get("id") or "")
+        return self._list_attachments(message_id)
+
     def fetch_reports(self, days: int = 7) -> Dict[str, Any]:
         """Fetch and ingest DMARC report attachments from Microsoft Graph."""
-        safe_days = max(1, min(int(days or 7), 365))
-        stats: Dict[str, Any] = {
-            "success": True,
-            "processed": 0,
-            "reports_found": 0,
-            "forensic_reports_found": 0,
-            "duplicate_reports": 0,
-            "duplicate_forensic_reports": 0,
-            "new_domains": [],
-            "errors": [],
-            "new_ingested_ids": [],
-            "details": [],
-            "target_mailbox": self._target_mailbox_label(),
-            "target_folder": self._target_folder_label(),
-            "search_window_days": safe_days,
-        }
+        safe_days = clamp_search_window(days)
+        stats = initial_import_stats(self.import_context(days=safe_days))
 
         try:
-            messages = self._list_dmarc_messages(days=safe_days)
+            messages = self.search_messages(days=safe_days)
         except Exception as exc:  # pylint: disable=broad-exception-caught
             logger.error("Microsoft Graph: failed to list messages: %s", exc)
-            return {**stats, "success": False, "error": str(exc), "errors": [str(exc)]}
+            return connector_failure_stats(
+                stats, "Failed to list Microsoft Graph messages.", error=exc
+            )
 
         domains_before = set(self.report_store.get_domains())
 
@@ -290,11 +300,7 @@ class MicrosoftGraphClient:
         return f"{LOGIN_BASE_URL}/{tenant}/oauth2/v2.0/token"
 
     def _append_detail(self, stats: dict, **detail: str) -> None:
-        detail.setdefault("mailbox", self._target_mailbox_label())
-        detail.setdefault("folder", self._target_folder_label())
-        stats.setdefault("details", []).append(
-            {key: value for key, value in detail.items() if value}
-        )
+        append_import_detail(stats, context=self.import_context(), **detail)
 
     def _target_mailbox_label(self) -> str:
         return self.mailbox or "authorized account"
@@ -339,11 +345,10 @@ class MicrosoftGraphClient:
             resp = httpx.request(method, url, headers=self._headers(), params=params, timeout=30)
             if resp.status_code == 401 and self.refresh_token:
                 self._refresh_access_token()
-                resp = httpx.request(method, url, headers=self._headers(), params=params, timeout=30)
-            if (
-                resp.status_code in _RETRYABLE_STATUS_CODES
-                and attempt < _MAX_GRAPH_RETRIES
-            ):
+                resp = httpx.request(
+                    method, url, headers=self._headers(), params=params, timeout=30
+                )
+            if resp.status_code in _RETRYABLE_STATUS_CODES and attempt < _MAX_GRAPH_RETRIES:
                 delay = self._retry_delay_seconds(resp, attempt)
                 logger.warning(
                     "Microsoft Graph request throttled/unavailable; retrying in %.1fs",
@@ -456,7 +461,11 @@ class MicrosoftGraphClient:
             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}")
+            stats["errors"].append(
+                sanitize_connector_error(
+                    f"Failed to fetch attachments for message {message_id}: {exc}"
+                )
+            )
             self._append_detail(
                 stats,
                 status="error",
@@ -562,7 +571,9 @@ class MicrosoftGraphClient:
                     )
             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}")
+                stats["errors"].append(
+                    sanitize_connector_error(f"Failed to parse {filename}: {exc}")
+                )
                 self._append_detail(
                     stats,
                     status="error",

backend/app/tests/test_mail_connector.py

@@ -0,0 +1,87 @@
+from app.services.mail_connector import (
+    ConnectorImportContext,
+    append_import_detail,
+    clamp_search_window,
+    connector_failure_stats,
+    dump_ingested_ids,
+    initial_import_stats,
+    load_ingested_ids,
+    sanitize_connector_error,
+)
+
+
+def test_initial_import_stats_includes_safe_context():
+    context = ConnectorImportContext(
+        source_type="M365_GRAPH",
+        mailbox="shared@example.com",
+        folder="DMARC Reports",
+        search_window_days=30,
+    )
+
+    stats = initial_import_stats(context)
+
+    assert stats["success"] is True
+    assert stats["source_type"] == "M365_GRAPH"
+    assert stats["target_mailbox"] == "shared@example.com"
+    assert stats["target_folder"] == "DMARC Reports"
+    assert stats["search_window_days"] == 30
+    assert stats["details"] == []
+
+
+def test_append_import_detail_redacts_secret_like_values():
+    stats = initial_import_stats()
+
+    append_import_detail(
+        stats,
+        status="error",
+        reason="provider_error",
+        error="access_token=abc123456789 client_secret=super-secret-value",
+    )
+
+    assert stats["details"] == [
+        {
+            "status": "error",
+            "reason": "provider_error",
+            "error": "access_token=**redacted** client_secret=**redacted**",
+        }
+    ]
+
+
+def test_connector_failure_stats_uses_redacted_error():
+    stats = initial_import_stats()
+    fake_token = "".join(["not", "-a-real", "-token"])
+
+    result = connector_failure_stats(
+        stats,
+        "Provider failed.",
+        error=f"Authorization: Bearer {fake_token}",
+    )
+
+    assert result["success"] is False
+    assert result["error"] == "Authorization: Bearer **redacted**"
+    assert result["errors"] == ["Authorization: Bearer **redacted**"]
+
+
+def test_ingested_id_helpers_normalize_values():
+    assert load_ingested_ids(None) == []
+    assert load_ingested_ids("{bad") == []
+    assert load_ingested_ids('{"not": "a-list"}') == []
+    assert load_ingested_ids('["a", 2]') == ["a", "2"]
+    assert dump_ingested_ids(["a", 2]) == '["a", "2"]'
+
+
+def test_clamp_search_window_bounds_values():
+    assert clamp_search_window(None) == 7
+    assert clamp_search_window("bad") == 7
+    assert clamp_search_window(0) == 7
+    assert clamp_search_window(-5) == 1
+    assert clamp_search_window(400) == 365
+
+
+def test_sanitize_connector_error_truncates_long_values():
+    value = "x" * 600
+
+    sanitized = sanitize_connector_error(value)
+
+    assert len(sanitized) == 500
+    assert sanitized.endswith("...")

docs/deployment/secrets.md

@@ -20,6 +20,8 @@ Store these values in a 1Password Environment for each deployment target:
 
 Non-sensitive values, such as `IMAP_SERVER`, `IMAP_USERNAME`, `LOGTO_ENDPOINT`, `LOGTO_APP_ID`, and `BACKEND_CORS_ORIGINS`, may also live in the Environment so each deployment has one complete configuration bundle.
 
+OAuth mail connectors can also store provider client secrets and refresh/access tokens in encrypted database fields after an administrator authorizes the source. Treat values such as `GMAIL_CLIENT_SECRET`, Microsoft 365 client secrets, refresh tokens, and access tokens as secrets even when they are provider-generated and short-lived.
+
 ## Create the Environment
 
 1. Open 1Password and enable the local MCP server or Environments feature if it is not already enabled.
@@ -89,6 +91,11 @@ Restrict the service user and file permissions so only the DMARQ process and the
 
 - Never commit `.env` files or secret values.
 - Never paste mailbox passwords, OAuth secrets, API tokens, database passwords, or generated session keys into issues, pull requests, logs, or chat.
+- Never include provider tokens, authorization headers, client secrets, raw mailbox payloads, or message bodies in connector diagnostics, import history, webhook payloads, screenshots, or support notes.
 - Keep `AUTH_DISABLED=true` limited to local development or a deployment protected by a separate authentication proxy.
 - Keep `LOGTO_SKIP_SSL_VERIFY=false` in production.
 - Use separate 1Password Environments for development, preprod, and production.
+
+## Connector Development
+
+New mail-source connectors must follow the shared connector contract in [Mail Connector Framework](../development/connectors.md). Use the shared sanitization helpers for provider errors and store only non-secret import context such as mailbox labels, folder labels, search windows, message IDs, filenames, domains, and report IDs.

docs/development/connectors.md

@@ -0,0 +1,56 @@
+# Mail Connector Framework
+
+DMARQ mailbox integrations should share one ingestion contract so new providers do not fork import behavior.
+
+## Connector Contract
+
+New mail-source connectors should implement the `MailSourceConnector` protocol in `backend/app/services/mail_connector.py`:
+
+- `import_context(days=None)` returns safe provider context such as source type, target mailbox, target folder, and search window. Do not include access tokens, refresh tokens, passwords, client secrets, authorization headers, raw provider payloads, or full message bodies.
+- `search_messages(days)` returns provider messages within the bounded search window.
+- `iter_attachments(message)` yields attachments for one provider message.
+- `fetch_reports(days=7)` runs the full ingestion path and returns the shared import-result shape.
+
+Use `ConnectorMessage` and `ConnectorAttachment` when provider data can be normalized cleanly. A connector may keep raw provider objects internally, but API responses and import history must use sanitized context and details only.
+
+## Import Result Shape
+
+Use `initial_import_stats()` to start an import result. The common keys are:
+
+- `success`
+- `processed`
+- `reports_found`
+- `forensic_reports_found`
+- `duplicate_reports`
+- `duplicate_forensic_reports`
+- `new_domains`
+- `errors`
+- `new_ingested_ids`
+- `details`
+
+Use `append_import_detail()` for message and attachment outcomes. Details should make retries understandable with reasons such as `already_ingested_message`, `unsupported_attachment`, `empty_attachment`, `parse_failed`, `duplicate`, or `imported`.
+
+Use `load_ingested_ids()` and `dump_ingested_ids()` for provider message IDs. A connector should mark a message as ingested only after the message was processed or determined to be safely skippable. Retryable message or attachment failures should not add the message ID to the ingested list.
+
+## Error Handling
+
+Provider failures must be mapped to sanitized, operator-readable diagnostics:
+
+- Use `sanitize_connector_error()` before storing or returning provider exception text.
+- Use `connector_failure_stats()` for failed list/search/setup paths.
+- Keep raw provider responses out of logs, import history, API responses, and frontend attributes.
+- Prefer bounded retries with provider backoff hints for throttling or temporary service failures.
+
+## Secret Handling
+
+Connectors may receive secrets from encrypted database fields or environment variables injected by the deployment runtime. They must not print or return those values.
+
+For local, preprod, and production deployments, prefer 1Password Environments or another runner-level secret injection mechanism. The connector code should only read the values it needs at runtime and should keep generated diagnostics safe for GitHub issues, import history, screenshots, and support requests.
+
+When adding a connector, add tests proving that:
+
+- duplicated provider message IDs do not inflate import totals,
+- parse failures and duplicates appear in `details`,
+- provider errors are redacted,
+- search/backfill windows are bounded,
+- secret-bearing strings are redacted before storage or API return.

docs/milestones.md

@@ -207,7 +207,7 @@ Planned:
 - Shared mailbox and folder selection support for DMARC report collection. Delivered with shared mailbox targeting, Microsoft Graph folder listing, folder-id based imports, UI selection, and mailbox/folder context in import history.
 - Import-history parity with existing sources (auditable attachment outcomes, duplicates, parse failures). Delivered for Microsoft 365 imports.
 - Backfill support with safe throttling and progressive search windows. Delivered with days-based Graph `receivedDateTime` filters, duplicate-safe reruns, and retry/backoff for throttled or temporarily unavailable Graph requests.
-- Secret handling mirrors existing guidance (no raw secrets in logs; 1Password-friendly).
+- Secret handling mirrors existing guidance (no raw secrets in logs; 1Password-friendly). Delivered with a shared connector protocol, sanitized import-result helpers, duplicate ID serialization helpers, and connector development guidance for future sources.
 
 Exit criteria:
 - A user can connect an Exchange Online mailbox, run an initial backfill, and then run scheduled polls with visible and trustworthy import history.