diff --git a/backend/alembic/versions/4e5f6a7b8c9d_add_workspace_operator_controls.py b/backend/alembic/versions/4e5f6a7b8c9d_add_workspace_operator_controls.py new file mode 100644 index 0000000..96a112d --- /dev/null +++ b/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") diff --git a/backend/app/api/api_v1/api.py b/backend/app/api/api_v1/api.py index fc02e01..27e5f9f 100644 --- a/backend/app/api/api_v1/api.py +++ b/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"]) diff --git a/backend/app/api/api_v1/endpoints/operator.py b/backend/app/api/api_v1/endpoints/operator.py new file mode 100644 index 0000000..3627c2a --- /dev/null +++ b/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), + } diff --git a/backend/app/models/workspace.py b/backend/app/models/workspace.py index 12e41fe..11cea11 100644 --- a/backend/app/models/workspace.py +++ b/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) diff --git a/backend/app/services/workspace_operator.py b/backend/app/services/workspace_operator.py new file mode 100644 index 0000000..5bcdaab --- /dev/null +++ b/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] diff --git a/backend/app/tests/test_workspace_operator.py b/backend/app/tests/test_workspace_operator.py new file mode 100644 index 0000000..2ab88fd --- /dev/null +++ b/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 "") diff --git a/docs/milestones.md b/docs/milestones.md index 62763b4..69f220d 100644 --- a/docs/milestones.md +++ b/docs/milestones.md @@ -261,7 +261,9 @@ Planned: - Templates for onboarding new workspaces. Delivered in M15.3: versioned workspace onboarding templates, preview/apply APIs, workspace/domain/mail source seeding, notification defaults, and operator validation checklists. -- Cross-workspace operator views for MSP admins, without weakening tenant isolation. +- Cross-workspace operator views for MSP admins. Delivered in M15.4: safe + workspace health summaries, recent drift detection, last-import/alert + rollups, and workspace retention controls without exposing tenant report rows. Exit criteria: - A single deployment can safely manage multiple client domains with clear boundaries and governance. diff --git a/docs/reference/api.md b/docs/reference/api.md index 328dc81..9b43c6b 100644 --- a/docs/reference/api.md +++ b/docs/reference/api.md @@ -174,6 +174,46 @@ the operator checklist, and writing a sanitized workspace audit event. Existing domains and mail sources are not duplicated. Existing notification settings are preserved unless `overwrite_existing` is set to `true`. +### MSP Operator Views + +#### List Workspace Operator Summaries + +```text +GET /operator/workspaces +``` + +Returns safe cross-workspace summaries for MSP operators: workspace identity, +health status, domain/mail-source counts, latest import status, active alert +count, recent drift event count, aggregate report count, and retention controls. +The endpoint does not return raw report records. + +#### Get One Workspace Summary + +```text +GET /operator/workspaces/{workspace_id} +``` + +Returns the same operator summary for a single workspace. + +#### Update Workspace Retention + +```text +PUT /operator/workspaces/{workspace_id}/retention +``` + +Request: + +```json +{ + "aggregate_reports_days": 730, + "forensic_reports_days": 120, + "tls_reports_days": 365 +} +``` + +Updates workspace retention controls and writes a sanitized +`workspace.retention_updated` audit event. + ### Domains #### List Domains diff --git a/docs/reference/database.md b/docs/reference/database.md index 5a722b2..3d9cf6e 100644 --- a/docs/reference/database.md +++ b/docs/reference/database.md @@ -25,6 +25,9 @@ workspace during migration. | name | VARCHAR | Display name | | description | TEXT | Optional operator-facing description | | active | BOOLEAN | Whether the workspace can be used | +| report_retention_days | INTEGER | Aggregate DMARC report retention target | +| forensic_retention_days | INTEGER | Forensic report retention target | +| tls_report_retention_days | INTEGER | SMTP TLS report retention target | | created_at | TIMESTAMP | When the workspace was created | | updated_at | TIMESTAMP | When the workspace was last updated | diff --git a/docs/reference/workspaces.md b/docs/reference/workspaces.md index eff3119..7ea0cf7 100644 --- a/docs/reference/workspaces.md +++ b/docs/reference/workspaces.md @@ -100,6 +100,36 @@ Notification defaults currently seed the existing notification settings table. They intentionally avoid Apprise target URLs, so operators still add and test delivery targets explicitly after onboarding. +## MSP Operator Views + +MSP operator endpoints provide cross-workspace summaries without returning raw +DMARC report rows across tenant boundaries. `GET /api/v1/operator/workspaces` +returns one summary per active workspace: + +- workspace identity and active state +- health status derived from domains, enabled mail sources, active alerts, + recent failed imports, and missing import history +- domain counts and names +- mail-source counts and the most recent import status +- aggregate report counts +- current retention controls +- recent workspace audit events as drift indicators + +`GET /api/v1/operator/workspaces/{workspace_id}` returns the same summary for +one workspace. + +Workspace retention controls are stored on the workspace row: + +| Field | Purpose | Default | +| --- | --- | --- | +| `report_retention_days` | Aggregate DMARC report retention target | 400 | +| `forensic_retention_days` | Forensic report retention target | 90 | +| `tls_report_retention_days` | SMTP TLS report retention target | 400 | + +Operators can update these controls with +`PUT /api/v1/operator/workspaces/{workspace_id}/retention`. Updates are written +to the workspace audit log as `workspace.retention_updated`. + The current implementation keeps domain names globally unique. That matches the existing single-domain ownership model and avoids ambiguous ownership while MSP RBAC and onboarding controls are built out.