feat: add MSP workspace operator views

backend/alembic/versions/4e5f6a7b8c9d_add_workspace_operator_controls.py

@@ -0,0 +1,37 @@
+"""add workspace operator controls
+
+Revision ID: 4e5f6a7b8c9d
+Revises: 3d4e5f6a7b8c
+Create Date: 2026-05-23 20:05:00.000000
+"""
+
+from typing import Sequence, Union
+
+import sqlalchemy as sa
+from alembic import op
+
+revision: str = "4e5f6a7b8c9d"
+down_revision: Union[str, None] = "3d4e5f6a7b8c"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    op.add_column(
+        "workspaces",
+        sa.Column("report_retention_days", sa.Integer(), nullable=False, server_default="400"),
+    )
+    op.add_column(
+        "workspaces",
+        sa.Column("forensic_retention_days", sa.Integer(), nullable=False, server_default="90"),
+    )
+    op.add_column(
+        "workspaces",
+        sa.Column("tls_report_retention_days", sa.Integer(), nullable=False, server_default="400"),
+    )
+
+
+def downgrade() -> None:
+    op.drop_column("workspaces", "tls_report_retention_days")
+    op.drop_column("workspaces", "forensic_retention_days")
+    op.drop_column("workspaces", "report_retention_days")

backend/app/api/api_v1/api.py

@@ -11,6 +11,7 @@ from app.api.api_v1.endpoints import (
     integrations,
     mail_sources,
     onboarding,
+    operator,
     public,
     reports,
     settings,
@@ -38,6 +39,7 @@ api_router.include_router(integrations.router, prefix="/integrations", tags=["in
 api_router.include_router(stats.router, prefix="/stats", tags=["stats"])
 api_router.include_router(mail_sources.router, prefix="/mail-sources", tags=["mail-sources"])
 api_router.include_router(onboarding.router, prefix="/onboarding", tags=["onboarding"])
+api_router.include_router(operator.router, prefix="/operator", tags=["operator"])
 api_router.include_router(settings.router, prefix="/settings", tags=["settings"])
 api_router.include_router(tls_reports.router, prefix="/tls-reports", tags=["tls-reports"])
 api_router.include_router(webhook.router, prefix="/webhook", tags=["webhook"])

backend/app/api/api_v1/endpoints/operator.py

@@ -0,0 +1,111 @@
+"""MSP operator endpoints."""
+
+from typing import Any, Dict, List
+
+from fastapi import APIRouter, Depends, HTTPException, Request, status
+from pydantic import BaseModel, Field
+from sqlalchemy.orm import Session
+
+from app.core.database import get_db
+from app.core.security import require_admin_auth
+from app.models.workspace import Workspace
+from app.services.workspace_access import (
+    PERMISSION_AUDIT_READ,
+    PERMISSION_WORKSPACE_ADMIN,
+    require_workspace_permission,
+)
+from app.services.workspace_audit import record_workspace_audit_log
+from app.services.workspace_operator import (
+    list_workspace_operator_summaries,
+    retention_to_dict,
+    workspace_operator_summary,
+)
+
+router = APIRouter()
+
+
+class OperatorWorkspacesResponse(BaseModel):
+    """Cross-workspace operator summaries."""
+
+    workspaces: List[Dict[str, Any]]
+
+
+class WorkspaceRetentionUpdate(BaseModel):
+    """Workspace retention controls."""
+
+    aggregate_reports_days: int = Field(..., ge=1, le=3650)
+    forensic_reports_days: int = Field(..., ge=1, le=3650)
+    tls_reports_days: int = Field(..., ge=1, le=3650)
+
+
+class WorkspaceRetentionResponse(BaseModel):
+    """Updated workspace retention response."""
+
+    workspace: Dict[str, Any]
+    retention: Dict[str, int]
+
+
+def _workspace_or_404(db: Session, workspace_id: int) -> Workspace:
+    workspace = db.query(Workspace).filter(Workspace.id == workspace_id).first()
+    if workspace is None:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail=f"Workspace {workspace_id} not found",
+        )
+    return workspace
+
+
+@router.get("/workspaces", response_model=OperatorWorkspacesResponse)
+async def list_operator_workspaces(
+    db: Session = Depends(get_db),
+    _auth: dict = Depends(require_admin_auth),
+) -> OperatorWorkspacesResponse:
+    """Return safe cross-workspace health, drift, import, and retention summaries."""
+    require_workspace_permission(_auth, PERMISSION_AUDIT_READ)
+    return {"workspaces": list_workspace_operator_summaries(db)}
+
+
+@router.get("/workspaces/{workspace_id}", response_model=Dict[str, Any])
+async def get_operator_workspace(
+    workspace_id: int,
+    db: Session = Depends(get_db),
+    _auth: dict = Depends(require_admin_auth),
+) -> Dict[str, Any]:
+    """Return one workspace operator summary."""
+    require_workspace_permission(_auth, PERMISSION_AUDIT_READ)
+    return workspace_operator_summary(db, _workspace_or_404(db, workspace_id))
+
+
+@router.put("/workspaces/{workspace_id}/retention", response_model=WorkspaceRetentionResponse)
+async def update_workspace_retention(
+    workspace_id: int,
+    payload: WorkspaceRetentionUpdate,
+    request: Request,
+    db: Session = Depends(get_db),
+    _auth: dict = Depends(require_admin_auth),
+) -> WorkspaceRetentionResponse:
+    """Update workspace retention controls and audit the change."""
+    require_workspace_permission(_auth, PERMISSION_WORKSPACE_ADMIN)
+    workspace = _workspace_or_404(db, workspace_id)
+    old_retention = retention_to_dict(workspace)
+    workspace.report_retention_days = payload.aggregate_reports_days
+    workspace.forensic_retention_days = payload.forensic_reports_days
+    workspace.tls_report_retention_days = payload.tls_reports_days
+    new_retention = retention_to_dict(workspace)
+    record_workspace_audit_log(
+        db,
+        workspace=workspace,
+        action="workspace.retention_updated",
+        entity_type="workspace",
+        entity_id=workspace.id,
+        entity_name=workspace.slug,
+        details={"old": old_retention, "new": new_retention},
+        auth_context=_auth,
+        request=request,
+    )
+    db.commit()
+    db.refresh(workspace)
+    return {
+        "workspace": {"id": workspace.id, "slug": workspace.slug, "name": workspace.name},
+        "retention": retention_to_dict(workspace),
+    }

backend/app/models/workspace.py

@@ -16,6 +16,9 @@ class Workspace(Base):
     name = Column(String, nullable=False)
     description = Column(Text, nullable=True)
     active = Column(Boolean, default=True, nullable=False, index=True)
+    report_retention_days = Column(Integer, default=400, nullable=False)
+    forensic_retention_days = Column(Integer, default=90, nullable=False)
+    tls_report_retention_days = Column(Integer, default=400, nullable=False)
     created_at = Column(DateTime, default=datetime.utcnow, index=True)
     updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
 

backend/app/services/workspace_operator.py

@@ -0,0 +1,189 @@
+"""Cross-workspace operator summaries for MSP mode."""
+
+# pylint: disable=not-callable
+
+from __future__ import annotations
+
+from datetime import datetime, timedelta
+from typing import Any, Dict, List, Optional
+
+from sqlalchemy import func
+from sqlalchemy.orm import Session
+
+from app.models.alert import AlertHistory
+from app.models.domain import Domain
+from app.models.mail_source import MailSource
+from app.models.mail_source_import import MailSourceImport
+from app.models.report import DMARCReport
+from app.models.workspace import Workspace
+from app.models.workspace_access import WorkspaceAuditLog
+
+
+def retention_to_dict(workspace: Workspace) -> Dict[str, int]:
+    """Return workspace retention controls."""
+    return {
+        "aggregate_reports_days": workspace.report_retention_days,
+        "forensic_reports_days": workspace.forensic_retention_days,
+        "tls_reports_days": workspace.tls_report_retention_days,
+    }
+
+
+def _last_import(db: Session, workspace: Workspace) -> Optional[MailSourceImport]:
+    return (
+        db.query(MailSourceImport)
+        .join(MailSource, MailSourceImport.mail_source_id == MailSource.id)
+        .filter(MailSource.workspace_id == workspace.id)
+        .order_by(MailSourceImport.finished_at.desc(), MailSourceImport.id.desc())
+        .first()
+    )
+
+
+def _recent_audit_rows(
+    db: Session,
+    workspace: Workspace,
+    *,
+    since: datetime,
+    limit: int = 5,
+) -> List[WorkspaceAuditLog]:
+    return (
+        db.query(WorkspaceAuditLog)
+        .filter(
+            WorkspaceAuditLog.workspace_id == workspace.id,
+            WorkspaceAuditLog.created_at >= since,
+        )
+        .order_by(WorkspaceAuditLog.created_at.desc(), WorkspaceAuditLog.id.desc())
+        .limit(limit)
+        .all()
+    )
+
+
+def _active_alert_count(db: Session, domain_names: List[str]) -> int:
+    if not domain_names:
+        return 0
+    return (
+        db.query(func.count(AlertHistory.id))
+        .filter(AlertHistory.domain.in_(domain_names), AlertHistory.is_active.is_(True))
+        .scalar()
+        or 0
+    )
+
+
+def _failed_import_count(db: Session, workspace: Workspace, since: datetime) -> int:
+    return (
+        db.query(func.count(MailSourceImport.id))
+        .join(MailSource, MailSourceImport.mail_source_id == MailSource.id)
+        .filter(
+            MailSource.workspace_id == workspace.id,
+            MailSourceImport.finished_at >= since,
+            MailSourceImport.status != "success",
+        )
+        .scalar()
+        or 0
+    )
+
+
+def _health_status(
+    *,
+    domain_count: int,
+    enabled_sources: int,
+    verified_domains: int,
+    active_alerts: int,
+    failed_imports: int,
+    last_import: Optional[MailSourceImport],
+) -> str:
+    if domain_count == 0 or enabled_sources == 0 or active_alerts > 0 or failed_imports > 0:
+        return "critical"
+    if verified_domains < domain_count or last_import is None:
+        return "warning"
+    return "healthy"
+
+
+def workspace_operator_summary(  # pylint: disable=too-many-locals
+    db: Session,
+    workspace: Workspace,
+    *,
+    now: Optional[datetime] = None,
+) -> Dict[str, Any]:
+    """Return one safe cross-workspace operator summary."""
+    now = now or datetime.utcnow()
+    since = now - timedelta(days=7)
+    domains = (
+        db.query(Domain).filter(Domain.workspace_id == workspace.id).order_by(Domain.name).all()
+    )
+    domain_names = [domain.name for domain in domains]
+    domain_count = len(domains)
+    active_domains = sum(1 for domain in domains if domain.active)
+    verified_domains = sum(1 for domain in domains if domain.verified)
+    source_count = (
+        db.query(func.count(MailSource.id)).filter(MailSource.workspace_id == workspace.id).scalar()
+        or 0
+    )
+    enabled_sources = (
+        db.query(func.count(MailSource.id))
+        .filter(MailSource.workspace_id == workspace.id, MailSource.enabled.is_(True))
+        .scalar()
+        or 0
+    )
+    report_count = (
+        db.query(func.count(DMARCReport.id))
+        .join(Domain, DMARCReport.domain_id == Domain.id)
+        .filter(Domain.workspace_id == workspace.id)
+        .scalar()
+        or 0
+    )
+    last_import = _last_import(db, workspace)
+    active_alerts = _active_alert_count(db, domain_names)
+    failed_imports = _failed_import_count(db, workspace, since)
+    drift_rows = _recent_audit_rows(db, workspace, since=since)
+    return {
+        "workspace": {
+            "id": workspace.id,
+            "slug": workspace.slug,
+            "name": workspace.name,
+            "active": workspace.active,
+        },
+        "health": {
+            "status": _health_status(
+                domain_count=domain_count,
+                enabled_sources=int(enabled_sources),
+                verified_domains=verified_domains,
+                active_alerts=active_alerts,
+                failed_imports=failed_imports,
+                last_import=last_import,
+            ),
+            "active_alerts": active_alerts,
+            "failed_imports_7d": failed_imports,
+            "drift_events_7d": len(drift_rows),
+        },
+        "domains": {
+            "total": domain_count,
+            "active": active_domains,
+            "verified": verified_domains,
+            "names": domain_names,
+        },
+        "mail_sources": {
+            "total": source_count,
+            "enabled": int(enabled_sources),
+            "last_import_at": last_import.finished_at.isoformat() if last_import else None,
+            "last_import_status": last_import.status if last_import else None,
+        },
+        "reports": {"aggregate_total": int(report_count)},
+        "retention": retention_to_dict(workspace),
+        "recent_drift": [
+            {
+                "action": row.action,
+                "entity_type": row.entity_type,
+                "entity_name": row.entity_name,
+                "created_at": row.created_at.isoformat() if row.created_at else None,
+            }
+            for row in drift_rows
+        ],
+    }
+
+
+def list_workspace_operator_summaries(db: Session) -> List[Dict[str, Any]]:
+    """Return safe summaries for every active workspace."""
+    workspaces = (
+        db.query(Workspace).filter(Workspace.active.is_(True)).order_by(Workspace.slug).all()
+    )
+    return [workspace_operator_summary(db, workspace) for workspace in workspaces]

backend/app/tests/test_workspace_operator.py

@@ -0,0 +1,161 @@
+from datetime import datetime, timedelta
+
+from fastapi.testclient import TestClient
+from sqlalchemy.orm import Session
+
+from app.models.alert import AlertHistory
+from app.models.domain import Domain
+from app.models.mail_source import MailSource
+from app.models.mail_source_import import MailSourceImport
+from app.models.report import DMARCReport
+from app.models.workspace import Workspace
+from app.models.workspace_access import WorkspaceAuditLog
+
+
+def _workspace(db_session: Session, slug: str, name: str) -> Workspace:
+    workspace = Workspace(slug=slug, name=name, active=True)
+    db_session.add(workspace)
+    db_session.flush()
+    return workspace
+
+
+def test_operator_view_summarizes_each_workspace(
+    authed_client: TestClient,
+    db_session: Session,
+):
+    """MSP operators can see health, import, alert, drift, and retention summaries."""
+    now = datetime.utcnow()
+    alpha = _workspace(db_session, "alpha", "Alpha Client")
+    beta = _workspace(db_session, "beta", "Beta Client")
+    alpha_domain = Domain(
+        workspace_id=alpha.id,
+        name="alpha.example",
+        active=True,
+        verified=True,
+    )
+    beta_domain = Domain(
+        workspace_id=beta.id,
+        name="beta.example",
+        active=True,
+        verified=False,
+    )
+    alpha_source = MailSource(
+        workspace_id=alpha.id,
+        name="Alpha inbox",
+        method="IMAP",
+        enabled=True,
+    )
+    beta_source = MailSource(
+        workspace_id=beta.id,
+        name="Beta inbox",
+        method="IMAP",
+        enabled=False,
+    )
+    db_session.add_all([alpha_domain, beta_domain, alpha_source, beta_source])
+    db_session.flush()
+    db_session.add_all(
+        [
+            DMARCReport(
+                domain_id=alpha_domain.id,
+                report_id="alpha-report",
+                org_name="Google",
+                begin_date=1,
+                end_date=2,
+            ),
+            MailSourceImport(
+                mail_source_id=alpha_source.id,
+                trigger="manual",
+                status="success",
+                finished_at=now,
+            ),
+            MailSourceImport(
+                mail_source_id=beta_source.id,
+                trigger="manual",
+                status="failed",
+                finished_at=now,
+            ),
+            AlertHistory(
+                fingerprint="beta-alert",
+                rule="missing_reports",
+                severity="warning",
+                domain="beta.example",
+                title="Missing reports",
+                detail="No reports received",
+                is_active=True,
+            ),
+            WorkspaceAuditLog(
+                workspace_id=beta.id,
+                actor_type="api_key",
+                action="mail_source.updated",
+                entity_type="mail_source",
+                entity_name="Beta inbox",
+                created_at=now - timedelta(days=1),
+            ),
+        ]
+    )
+    db_session.commit()
+
+    response = authed_client.get("/api/v1/operator/workspaces")
+
+    assert response.status_code == 200
+    by_slug = {item["workspace"]["slug"]: item for item in response.json()["workspaces"]}
+    assert by_slug["alpha"]["health"]["status"] == "healthy"
+    assert by_slug["alpha"]["reports"]["aggregate_total"] == 1
+    assert by_slug["alpha"]["mail_sources"]["last_import_status"] == "success"
+    assert by_slug["beta"]["health"]["status"] == "critical"
+    assert by_slug["beta"]["health"]["active_alerts"] == 1
+    assert by_slug["beta"]["health"]["failed_imports_7d"] == 1
+    assert by_slug["beta"]["health"]["drift_events_7d"] == 1
+    assert by_slug["beta"]["recent_drift"][0]["action"] == "mail_source.updated"
+
+
+def test_operator_view_does_not_include_inactive_workspaces(
+    authed_client: TestClient,
+    db_session: Session,
+):
+    """Inactive workspaces are not listed in the cross-workspace operator view."""
+    _workspace(db_session, "active", "Active Client")
+    inactive = Workspace(slug="inactive", name="Inactive Client", active=False)
+    db_session.add(inactive)
+    db_session.commit()
+
+    response = authed_client.get("/api/v1/operator/workspaces")
+
+    assert response.status_code == 200
+    slugs = {item["workspace"]["slug"] for item in response.json()["workspaces"]}
+    assert slugs == {"active"}
+
+
+def test_workspace_retention_update_is_audited(
+    authed_client: TestClient,
+    db_session: Session,
+):
+    """Operators can update workspace retention controls with an audit record."""
+    workspace = _workspace(db_session, "client", "Client")
+    db_session.commit()
+
+    response = authed_client.put(
+        f"/api/v1/operator/workspaces/{workspace.id}/retention",
+        json={
+            "aggregate_reports_days": 730,
+            "forensic_reports_days": 120,
+            "tls_reports_days": 365,
+        },
+        headers={"x-forwarded-for": "203.0.113.9"},
+    )
+
+    assert response.status_code == 200
+    db_session.refresh(workspace)
+    assert workspace.report_retention_days == 730
+    assert workspace.forensic_retention_days == 120
+    assert workspace.tls_report_retention_days == 365
+    assert response.json()["retention"]["aggregate_reports_days"] == 730
+
+    audit = (
+        db_session.query(WorkspaceAuditLog)
+        .filter(WorkspaceAuditLog.action == "workspace.retention_updated")
+        .one()
+    )
+    assert audit.workspace_id == workspace.id
+    assert audit.ip_address == "203.0.113.9"
+    assert "730" in (audit.details or "")