Merge pull request #184 from christianlouis/codex/m12-connector-framework

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("...")