from __future__ import annotations from fastapi import APIRouter, Depends, HTTPException, Query, status from govoplan_mail.backend.schemas import ( MailConnectionTestResponse, MailImapFolderListResponse, MailImapFolderResponse, MailImapTestRequest, MailMailboxAttachmentResponse, MailMailboxMessageDetailResponse, MailMailboxMessageListResponse, MailMailboxMessageResponse, MailMailboxMessageSummaryResponse, MailProfilePolicyResponse, MailProfilePolicyUpdateRequest, MailServerProfileCreateRequest, MailServerProfileListResponse, MailServerProfileResponse, MailServerProfileUpdateRequest, MailSmtpTestRequest, ) from govoplan_access.backend.auth.dependencies import ApiPrincipal, get_api_principal, has_scope from govoplan_core.db.session import get_session from govoplan_mail.backend.mail_profiles import ( MailProfileError, create_mail_server_profile, effective_mail_profile_policy_for_scope, get_mail_profile_policy, get_mail_server_profile, imap_config_from_profile, list_mail_server_profiles, parent_mail_profile_policy, profile_response_payload, set_mail_profile_policy, smtp_config_from_profile, update_mail_server_profile, ) from govoplan_mail.backend.config import ImapConfig, SmtpConfig from govoplan_mail.backend.sending.imap import ImapAppendError, ImapConfigurationError, get_imap_message, list_imap_folders, list_imap_messages, test_imap_login from govoplan_mail.backend.sending.smtp import test_smtp_login from sqlalchemy.orm import Session router = APIRouter(prefix="/mail", tags=["mail"]) def _require_scope(principal: ApiPrincipal, scope: str) -> None: if not has_scope(principal, scope): raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail=f"Missing scope: {scope}") def _require_any_scope(principal: ApiPrincipal, *scopes: str) -> None: if not any(has_scope(principal, scope) for scope in scopes): raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Requires one of: " + ", ".join(scopes)) def _require_profile_write_scope(principal: ApiPrincipal, scope_type: str) -> None: if scope_type == "system": _require_scope(principal, "system:settings:write") else: _require_scope(principal, "mail:profile:write") def _require_profile_credentials_scope(principal: ApiPrincipal, scope_type: str) -> None: if scope_type == "system": _require_scope(principal, "system:settings:write") else: _require_scope(principal, "mail:secret:manage") def _require_mailbox_read_scope(principal: ApiPrincipal) -> None: _require_scope(principal, "mail:mailbox:read") _require_scope(principal, "mail:profile:use") def _require_policy_read_scope(principal: ApiPrincipal, scope_type: str) -> None: if scope_type == "system": _require_scope(principal, "system:settings:read") elif scope_type == "tenant": _require_any_scope(principal, "admin:policies:read", "mail:profile:read") else: _require_any_scope(principal, "admin:policies:read", "mail:profile:read", "campaigns:campaign:read") def _require_policy_write_scope(principal: ApiPrincipal, scope_type: str) -> None: if scope_type == "system": _require_scope(principal, "system:settings:write") elif scope_type == "tenant": _require_scope(principal, "admin:policies:write") else: _require_any_scope(principal, "admin:policies:write", "mail:profile:write") def _profile_response(profile) -> MailServerProfileResponse: return MailServerProfileResponse.model_validate(profile_response_payload(profile)) def _mailbox_summary_response(message) -> MailMailboxMessageSummaryResponse: return MailMailboxMessageSummaryResponse( uid=message.uid, folder=message.folder, subject=message.subject, from_header=message.from_header, to_header=message.to_header, cc_header=message.cc_header, date=message.date, message_id=message.message_id, flags=message.flags, size_bytes=message.size_bytes, body_preview=message.body_preview, attachment_count=message.attachment_count, ) def _mailbox_detail_response(message) -> MailMailboxMessageDetailResponse: return MailMailboxMessageDetailResponse( uid=message.uid, folder=message.folder, subject=message.subject, from_header=message.from_header, to_header=message.to_header, cc_header=message.cc_header, date=message.date, message_id=message.message_id, flags=message.flags, size_bytes=message.size_bytes, body_preview=message.body_preview, body_text=message.body_text, body_html=message.body_html, headers=message.headers, attachments=[MailMailboxAttachmentResponse(filename=item.filename, content_type=item.content_type, size_bytes=item.size_bytes) for item in message.attachments], ) def _imap_config_for_profile(session: Session, *, tenant_id: str, profile_id: str): profile = get_mail_server_profile(session, tenant_id=tenant_id, profile_id=profile_id, require_active=True) imap = imap_config_from_profile(profile) if imap is None: raise MailProfileError("Mail-server profile has no IMAP configuration") return imap def _parent_mail_profile_policy_payload(session: Session, *, tenant_id: str, scope_type: str, scope_id: str | None): if scope_type == "system": return None return parent_mail_profile_policy(session, tenant_id=tenant_id, scope_type=scope_type, scope_id=scope_id).as_dict() @router.get("/profiles", response_model=MailServerProfileListResponse) def list_profiles( include_inactive: bool = False, campaign_id: str | None = Query(default=None), principal: ApiPrincipal = Depends(get_api_principal), session: Session = Depends(get_session), ): _require_any_scope(principal, "mail:profile:read", "system:settings:read") try: profiles = list_mail_server_profiles( session, tenant_id=principal.tenant_id, include_inactive=include_inactive, campaign_id=campaign_id, ) return MailServerProfileListResponse(profiles=[_profile_response(profile) for profile in profiles]) except MailProfileError as exc: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(exc)) from exc @router.post("/profiles", response_model=MailServerProfileResponse, status_code=status.HTTP_201_CREATED) def create_profile( payload: MailServerProfileCreateRequest, principal: ApiPrincipal = Depends(get_api_principal), session: Session = Depends(get_session), ): _require_profile_write_scope(principal, payload.scope_type) if payload.credentials_supplied(): _require_profile_credentials_scope(principal, payload.scope_type) try: profile = create_mail_server_profile( session, tenant_id=principal.tenant_id, user_id=principal.user.id, name=payload.name, slug=payload.slug, description=payload.description, smtp=payload.smtp_config(), imap=payload.imap_config(), is_active=payload.is_active, scope_type=payload.scope_type, scope_id=payload.scope_id, ) session.commit() session.refresh(profile) return _profile_response(profile) except MailProfileError as exc: session.rollback() raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail=str(exc)) from exc @router.get("/profiles/{profile_id}", response_model=MailServerProfileResponse) def get_profile( profile_id: str, principal: ApiPrincipal = Depends(get_api_principal), session: Session = Depends(get_session), ): _require_any_scope(principal, "mail:profile:read", "system:settings:read") try: return _profile_response(get_mail_server_profile(session, tenant_id=principal.tenant_id, profile_id=profile_id)) except MailProfileError as exc: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(exc)) from exc @router.patch("/profiles/{profile_id}", response_model=MailServerProfileResponse) def update_profile( profile_id: str, payload: MailServerProfileUpdateRequest, principal: ApiPrincipal = Depends(get_api_principal), session: Session = Depends(get_session), ): try: profile = get_mail_server_profile(session, tenant_id=principal.tenant_id, profile_id=profile_id) _require_profile_write_scope(principal, profile.scope_type or "tenant") if payload.credentials_supplied() or payload.clear_imap: _require_profile_credentials_scope(principal, profile.scope_type or "tenant") smtp_config = payload.smtp_config() if payload.smtp is None and "smtp" in payload.credentials.model_fields_set: smtp_data = dict(profile.smtp_config or {}) smtp_data.update(payload.credentials.smtp.model_dump(mode="json", exclude_unset=True)) smtp_config = SmtpConfig.model_validate(smtp_data) imap_config = payload.imap_config() if payload.imap is None and "imap" in payload.credentials.model_fields_set and profile.imap_config: imap_data = dict(profile.imap_config or {}) imap_data.update(payload.credentials.imap.model_dump(mode="json", exclude_unset=True)) imap_config = ImapConfig.model_validate(imap_data) update_mail_server_profile( session, profile, tenant_id=principal.tenant_id, user_id=principal.user.id, name=payload.name, slug=payload.slug, description=payload.description if "description" in payload.model_fields_set else None, is_active=payload.is_active, smtp=smtp_config, imap=imap_config, clear_imap=payload.clear_imap, ) session.commit() session.refresh(profile) return _profile_response(profile) except MailProfileError as exc: session.rollback() raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail=str(exc)) from exc @router.delete("/profiles/{profile_id}", response_model=MailServerProfileResponse) def deactivate_profile( profile_id: str, principal: ApiPrincipal = Depends(get_api_principal), session: Session = Depends(get_session), ): try: profile = get_mail_server_profile(session, tenant_id=principal.tenant_id, profile_id=profile_id) _require_profile_write_scope(principal, profile.scope_type or "tenant") profile.is_active = False session.add(profile) session.commit() session.refresh(profile) return _profile_response(profile) except MailProfileError as exc: session.rollback() raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(exc)) from exc @router.get("/policies/{scope_type}", response_model=MailProfilePolicyResponse) def read_mail_profile_policy( scope_type: str, scope_id: str | None = Query(default=None), campaign_id: str | None = Query(default=None), principal: ApiPrincipal = Depends(get_api_principal), session: Session = Depends(get_session), ): scope_type = scope_type.strip().casefold() _require_policy_read_scope(principal, scope_type) try: policy = get_mail_profile_policy(session, tenant_id=principal.tenant_id, scope_type=scope_type, scope_id=scope_id) effective_policy = effective_mail_profile_policy_for_scope( session, tenant_id=principal.tenant_id, scope_type=scope_type, scope_id=scope_id, campaign_id=campaign_id, ) effective = effective_policy.as_dict() effective_sources = effective_policy.source_policies parent_policy = parent_mail_profile_policy(session, tenant_id=principal.tenant_id, scope_type=scope_type, scope_id=scope_id) if scope_type != "system" else None parent = parent_policy.as_dict() if parent_policy else None parent_sources = parent_policy.source_policies if parent_policy else [] return MailProfilePolicyResponse(scope_type=scope_type, scope_id=scope_id, policy=policy, effective_policy=effective, parent_policy=parent, effective_policy_sources=effective_sources, parent_policy_sources=parent_sources) except MailProfileError as exc: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(exc)) from exc @router.put("/policies/{scope_type}", response_model=MailProfilePolicyResponse) def write_mail_profile_policy( scope_type: str, payload: MailProfilePolicyUpdateRequest, scope_id: str | None = Query(default=None), principal: ApiPrincipal = Depends(get_api_principal), session: Session = Depends(get_session), ): scope_type = scope_type.strip().casefold() _require_policy_write_scope(principal, scope_type) try: policy = set_mail_profile_policy( session, tenant_id=principal.tenant_id, scope_type=scope_type, scope_id=scope_id, policy=payload.policy.model_dump(mode="json"), ) session.commit() effective_policy = effective_mail_profile_policy_for_scope( session, tenant_id=principal.tenant_id, scope_type=scope_type, scope_id=scope_id, ) effective = effective_policy.as_dict() effective_sources = effective_policy.source_policies parent_policy = parent_mail_profile_policy(session, tenant_id=principal.tenant_id, scope_type=scope_type, scope_id=scope_id) if scope_type != "system" else None parent = parent_policy.as_dict() if parent_policy else None parent_sources = parent_policy.source_policies if parent_policy else [] return MailProfilePolicyResponse(scope_type=scope_type, scope_id=scope_id, policy=policy, effective_policy=effective, parent_policy=parent, effective_policy_sources=effective_sources, parent_policy_sources=parent_sources) except MailProfileError as exc: session.rollback() raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail=str(exc)) from exc @router.post("/profiles/{profile_id}/test-smtp", response_model=MailConnectionTestResponse) def test_profile_smtp( profile_id: str, principal: ApiPrincipal = Depends(get_api_principal), session: Session = Depends(get_session), ): _require_scope(principal, "mail:profile:test") _require_scope(principal, "mail:profile:use") try: profile = get_mail_server_profile(session, tenant_id=principal.tenant_id, profile_id=profile_id, require_active=True) smtp = smtp_config_from_profile(profile) result = test_smtp_login(smtp_config=smtp) return MailConnectionTestResponse(ok=True, protocol="smtp", host=result.host, port=result.port, security=result.security, message="SMTP connection successful.", details={"authenticated": result.authenticated}) except MailProfileError as exc: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(exc)) from exc except Exception as exc: return MailConnectionTestResponse(ok=False, protocol="smtp", message=_safe_error_message(exc), details={"error_type": exc.__class__.__name__}) @router.post("/profiles/{profile_id}/test-imap", response_model=MailConnectionTestResponse) def test_profile_imap( profile_id: str, principal: ApiPrincipal = Depends(get_api_principal), session: Session = Depends(get_session), ): _require_scope(principal, "mail:profile:test") _require_scope(principal, "mail:profile:use") try: profile = get_mail_server_profile(session, tenant_id=principal.tenant_id, profile_id=profile_id, require_active=True) imap = imap_config_from_profile(profile) if imap is None: raise MailProfileError("Mail-server profile has no IMAP configuration") result = test_imap_login(imap_config=imap) return MailConnectionTestResponse(ok=True, protocol="imap", host=result.host, port=result.port, security=result.security, message="IMAP connection successful.", details={"authenticated": result.authenticated}) except MailProfileError as exc: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(exc)) from exc except Exception as exc: return MailConnectionTestResponse(ok=False, protocol="imap", message=_safe_error_message(exc), details={"error_type": exc.__class__.__name__}) @router.post("/profiles/{profile_id}/list-imap-folders", response_model=MailImapFolderListResponse) def list_profile_imap_folders( profile_id: str, principal: ApiPrincipal = Depends(get_api_principal), session: Session = Depends(get_session), ): _require_scope(principal, "mail:profile:test") _require_scope(principal, "mail:profile:use") try: profile = get_mail_server_profile(session, tenant_id=principal.tenant_id, profile_id=profile_id, require_active=True) imap = imap_config_from_profile(profile) if imap is None: raise MailProfileError("Mail-server profile has no IMAP configuration") result = list_imap_folders(imap_config=imap) folders = [MailImapFolderResponse(name=item.name, flags=item.flags, message_count=item.message_count, unseen_count=item.unseen_count) for item in result.folders] return MailImapFolderListResponse(ok=True, host=result.host, port=result.port, security=result.security, message=f"Found {len(folders)} IMAP folder(s).", folders=folders, detected_sent_folder=result.detected_sent_folder) except MailProfileError as exc: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(exc)) from exc except Exception as exc: return MailImapFolderListResponse(ok=False, message=_safe_error_message(exc), folders=[], detected_sent_folder=None, details={"error_type": exc.__class__.__name__}) @router.get("/profiles/{profile_id}/mailbox/folders", response_model=MailImapFolderListResponse) def list_profile_mailbox_folders( profile_id: str, principal: ApiPrincipal = Depends(get_api_principal), session: Session = Depends(get_session), ): _require_mailbox_read_scope(principal) try: imap = _imap_config_for_profile(session, tenant_id=principal.tenant_id, profile_id=profile_id) result = list_imap_folders(imap_config=imap) folders = [MailImapFolderResponse(name=item.name, flags=item.flags, message_count=item.message_count, unseen_count=item.unseen_count) for item in result.folders] return MailImapFolderListResponse(ok=True, host=result.host, port=result.port, security=result.security, message=f"Found {len(folders)} IMAP folder(s).", folders=folders, detected_sent_folder=result.detected_sent_folder) except MailProfileError as exc: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(exc)) from exc except ImapConfigurationError as exc: raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail=str(exc)) from exc except ImapAppendError as exc: raise HTTPException(status_code=status.HTTP_502_BAD_GATEWAY, detail=str(exc)) from exc @router.get("/profiles/{profile_id}/mailbox/messages", response_model=MailMailboxMessageListResponse) def list_profile_mailbox_messages( profile_id: str, folder: str = Query(default="INBOX", min_length=1, max_length=255), limit: int = Query(default=50, ge=1, le=100), offset: int = Query(default=0, ge=0, le=100000), principal: ApiPrincipal = Depends(get_api_principal), session: Session = Depends(get_session), ): _require_mailbox_read_scope(principal) try: imap = _imap_config_for_profile(session, tenant_id=principal.tenant_id, profile_id=profile_id) result = list_imap_messages(imap_config=imap, folder=folder, limit=limit, offset=offset) return MailMailboxMessageListResponse( profile_id=profile_id, folder=result.folder, host=result.host, port=result.port, security=result.security, total_count=result.total_count, offset=result.offset, limit=result.limit, messages=[_mailbox_summary_response(message) for message in result.messages], ) except MailProfileError as exc: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(exc)) from exc except ImapConfigurationError as exc: raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail=str(exc)) from exc except ImapAppendError as exc: raise HTTPException(status_code=status.HTTP_502_BAD_GATEWAY, detail=str(exc)) from exc @router.get("/profiles/{profile_id}/mailbox/messages/{message_uid}", response_model=MailMailboxMessageResponse) def get_profile_mailbox_message( profile_id: str, message_uid: str, folder: str = Query(default="INBOX", min_length=1, max_length=255), principal: ApiPrincipal = Depends(get_api_principal), session: Session = Depends(get_session), ): _require_mailbox_read_scope(principal) try: imap = _imap_config_for_profile(session, tenant_id=principal.tenant_id, profile_id=profile_id) result = get_imap_message(imap_config=imap, folder=folder, uid=message_uid) return MailMailboxMessageResponse( profile_id=profile_id, folder=result.folder, host=result.host, port=result.port, security=result.security, message=_mailbox_detail_response(result.message), ) except MailProfileError as exc: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(exc)) from exc except ImapConfigurationError as exc: raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail=str(exc)) from exc except ImapAppendError as exc: status_code = status.HTTP_404_NOT_FOUND if "not found" in str(exc).casefold() else status.HTTP_502_BAD_GATEWAY raise HTTPException(status_code=status_code, detail=str(exc)) from exc def _safe_error_message(exc: Exception) -> str: text = str(exc).strip() return text or exc.__class__.__name__ @router.post("/test-smtp", response_model=MailConnectionTestResponse) def test_smtp_settings( payload: MailSmtpTestRequest, principal: ApiPrincipal = Depends(get_api_principal), ): """Test SMTP connectivity/login without sending any message.""" if not has_scope(principal, "mail:profile:test"): raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Missing scope: mail_servers:test") if not has_scope(principal, "mail:secret:manage"): raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Testing raw credentials requires mail_servers:manage_credentials") try: result = test_smtp_login(smtp_config=payload) return MailConnectionTestResponse( ok=True, protocol="smtp", host=result.host, port=result.port, security=result.security, message="SMTP connection successful.", details={"authenticated": result.authenticated}, ) except Exception as exc: return MailConnectionTestResponse( ok=False, protocol="smtp", host=payload.host, port=payload.port, security=payload.security.value, message=_safe_error_message(exc), details={"error_type": exc.__class__.__name__}, ) @router.post("/test-imap", response_model=MailConnectionTestResponse) def test_imap_settings( payload: MailImapTestRequest, principal: ApiPrincipal = Depends(get_api_principal), ): """Test IMAP connectivity/login without selecting or appending messages.""" if not has_scope(principal, "mail:profile:test"): raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Missing scope: mail_servers:test") if not has_scope(principal, "mail:secret:manage"): raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Testing raw credentials requires mail_servers:manage_credentials") try: result = test_imap_login(imap_config=payload) return MailConnectionTestResponse( ok=True, protocol="imap", host=result.host, port=result.port, security=result.security, message="IMAP connection successful.", details={"authenticated": result.authenticated}, ) except Exception as exc: return MailConnectionTestResponse( ok=False, protocol="imap", host=payload.host, port=payload.port, security=payload.security.value, message=_safe_error_message(exc), details={"error_type": exc.__class__.__name__}, ) @router.post("/list-imap-folders", response_model=MailImapFolderListResponse) def list_imap_folder_settings( payload: MailImapTestRequest, principal: ApiPrincipal = Depends(get_api_principal), ): """List visible IMAP folders and return the best Sent-folder guess.""" if not has_scope(principal, "mail:profile:test"): raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Missing scope: mail_servers:test") if not has_scope(principal, "mail:secret:manage"): raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Testing raw credentials requires mail_servers:manage_credentials") try: result = list_imap_folders(imap_config=payload) folders = [MailImapFolderResponse(name=item.name, flags=item.flags, message_count=item.message_count, unseen_count=item.unseen_count) for item in result.folders] return MailImapFolderListResponse( ok=True, host=result.host, port=result.port, security=result.security, message=f"Found {len(folders)} IMAP folder(s).", folders=folders, detected_sent_folder=result.detected_sent_folder, ) except Exception as exc: return MailImapFolderListResponse( ok=False, host=payload.host, port=payload.port, security=payload.security.value, message=_safe_error_message(exc), folders=[], detected_sent_folder=None, details={"error_type": exc.__class__.__name__}, )