295 lines
10 KiB
Python
295 lines
10 KiB
Python
from __future__ import annotations
|
|
|
|
from datetime import datetime
|
|
from typing import Any, Literal
|
|
|
|
from pydantic import BaseModel, ConfigDict, Field, model_validator
|
|
|
|
from govoplan_core.api.v1.schemas import DeltaDeletedItem
|
|
from govoplan_mail.backend.config import ImapConfig, ImapServerConfig, SmtpConfig, SmtpServerConfig, TransportCredentials
|
|
|
|
|
|
class MailSmtpTestRequest(SmtpConfig):
|
|
"""SMTP settings supplied directly from the WebUI mail settings form."""
|
|
|
|
|
|
class MailImapTestRequest(ImapConfig):
|
|
"""IMAP settings supplied directly from the WebUI mail settings form."""
|
|
|
|
|
|
class MailTransportCredentialsPayload(TransportCredentials):
|
|
"""Credentials supplied separately from saved server settings."""
|
|
|
|
|
|
class MailServerProfileCredentialsPayload(BaseModel):
|
|
model_config = ConfigDict(extra="forbid")
|
|
|
|
smtp: MailTransportCredentialsPayload = Field(default_factory=MailTransportCredentialsPayload)
|
|
imap: MailTransportCredentialsPayload = Field(default_factory=MailTransportCredentialsPayload)
|
|
|
|
|
|
def _normalize_profile_transport_payload(value: object) -> object:
|
|
if not isinstance(value, dict):
|
|
return value
|
|
data = dict(value)
|
|
credentials = data.get("credentials") if isinstance(data.get("credentials"), dict) else {}
|
|
credentials = {key: dict(item) for key, item in credentials.items() if isinstance(item, dict)}
|
|
for protocol in ("smtp", "imap"):
|
|
transport = data.get(protocol)
|
|
if not isinstance(transport, dict):
|
|
continue
|
|
next_transport = dict(transport)
|
|
next_credentials = dict(credentials.get(protocol) or {})
|
|
for field in ("username", "password"):
|
|
if field in next_transport and field not in next_credentials:
|
|
next_credentials[field] = next_transport[field]
|
|
next_transport.pop(field, None)
|
|
next_transport.pop("enabled", None)
|
|
data[protocol] = next_transport
|
|
if next_credentials:
|
|
credentials[protocol] = next_credentials
|
|
if credentials:
|
|
data["credentials"] = credentials
|
|
return data
|
|
|
|
|
|
def _merge_transport_credentials(server: SmtpServerConfig | ImapServerConfig, credentials: MailTransportCredentialsPayload) -> dict[str, object]:
|
|
payload = server.model_dump(mode="json")
|
|
if "username" in credentials.model_fields_set:
|
|
payload["username"] = credentials.username
|
|
if "password" in credentials.model_fields_set:
|
|
payload["password"] = credentials.password
|
|
return payload
|
|
|
|
|
|
MailProfileScope = Literal["system", "tenant", "user", "group", "campaign"]
|
|
|
|
|
|
class MailCredentialPolicyPayload(BaseModel):
|
|
model_config = ConfigDict(extra="forbid")
|
|
|
|
inherit: bool | None = None
|
|
|
|
|
|
class MailProfilePolicyPayload(BaseModel):
|
|
model_config = ConfigDict(extra="forbid")
|
|
|
|
allowed_profile_ids: list[str] = Field(default_factory=list)
|
|
allow_user_profiles: bool | None = None
|
|
allow_group_profiles: bool | None = None
|
|
allow_campaign_profiles: bool | None = None
|
|
smtp_credentials: MailCredentialPolicyPayload = Field(default_factory=MailCredentialPolicyPayload)
|
|
imap_credentials: MailCredentialPolicyPayload = Field(default_factory=MailCredentialPolicyPayload)
|
|
whitelist: dict[str, list[str]] = Field(default_factory=dict)
|
|
blacklist: dict[str, list[str]] = Field(default_factory=dict)
|
|
allow_lower_level_limits: dict[str, bool] = Field(default_factory=dict)
|
|
|
|
|
|
class MailProfilePolicyUpdateRequest(BaseModel):
|
|
model_config = ConfigDict(extra="forbid")
|
|
|
|
policy: MailProfilePolicyPayload = Field(default_factory=MailProfilePolicyPayload)
|
|
|
|
|
|
class PolicySourceStepResponse(BaseModel):
|
|
scope_type: str
|
|
scope_id: str | None = None
|
|
path: str
|
|
label: str
|
|
applied_fields: list[str] = Field(default_factory=list)
|
|
policy: dict[str, Any] = Field(default_factory=dict)
|
|
|
|
|
|
class MailProfilePolicyResponse(BaseModel):
|
|
scope_type: MailProfileScope
|
|
scope_id: str | None = None
|
|
policy: dict[str, Any]
|
|
effective_policy: dict[str, Any] | None = None
|
|
parent_policy: dict[str, Any] | None = None
|
|
effective_policy_sources: list[PolicySourceStepResponse] = Field(default_factory=list)
|
|
parent_policy_sources: list[PolicySourceStepResponse] = Field(default_factory=list)
|
|
|
|
|
|
class MailServerProfileCreateRequest(BaseModel):
|
|
model_config = ConfigDict(extra="forbid")
|
|
|
|
name: str = Field(min_length=1, max_length=255)
|
|
slug: str | None = Field(default=None, max_length=100)
|
|
description: str | None = None
|
|
is_active: bool = True
|
|
scope_type: MailProfileScope = "tenant"
|
|
scope_id: str | None = None
|
|
smtp: SmtpServerConfig
|
|
imap: ImapServerConfig | None = None
|
|
credentials: MailServerProfileCredentialsPayload = Field(default_factory=MailServerProfileCredentialsPayload)
|
|
|
|
@model_validator(mode="before")
|
|
@classmethod
|
|
def normalize_legacy_transport_payload(cls, value: object) -> object:
|
|
return _normalize_profile_transport_payload(value)
|
|
|
|
def smtp_config(self) -> SmtpConfig:
|
|
return SmtpConfig.model_validate(_merge_transport_credentials(self.smtp, self.credentials.smtp))
|
|
|
|
def imap_config(self) -> ImapConfig | None:
|
|
if self.imap is None:
|
|
return None
|
|
return ImapConfig.model_validate(_merge_transport_credentials(self.imap, self.credentials.imap))
|
|
|
|
def credentials_supplied(self) -> bool:
|
|
return any(
|
|
field in credentials.model_fields_set
|
|
for credentials in (self.credentials.smtp, self.credentials.imap)
|
|
for field in ("username", "password")
|
|
)
|
|
|
|
|
|
class MailServerProfileUpdateRequest(BaseModel):
|
|
model_config = ConfigDict(extra="forbid")
|
|
|
|
name: str | None = Field(default=None, max_length=255)
|
|
slug: str | None = Field(default=None, max_length=100)
|
|
description: str | None = None
|
|
is_active: bool | None = None
|
|
smtp: SmtpServerConfig | None = None
|
|
imap: ImapServerConfig | None = None
|
|
credentials: MailServerProfileCredentialsPayload = Field(default_factory=MailServerProfileCredentialsPayload)
|
|
clear_imap: bool = False
|
|
|
|
@model_validator(mode="before")
|
|
@classmethod
|
|
def normalize_legacy_transport_payload(cls, value: object) -> object:
|
|
return _normalize_profile_transport_payload(value)
|
|
|
|
def smtp_config(self) -> SmtpConfig | None:
|
|
if self.smtp is None and "smtp" not in self.credentials.model_fields_set:
|
|
return None
|
|
server = self.smtp or SmtpServerConfig()
|
|
return SmtpConfig.model_validate(_merge_transport_credentials(server, self.credentials.smtp))
|
|
|
|
def imap_config(self) -> ImapConfig | None:
|
|
if self.imap is None and "imap" not in self.credentials.model_fields_set:
|
|
return None
|
|
server = self.imap or ImapServerConfig()
|
|
return ImapConfig.model_validate(_merge_transport_credentials(server, self.credentials.imap))
|
|
|
|
def credentials_supplied(self) -> bool:
|
|
return any(
|
|
field in credentials.model_fields_set
|
|
for credentials in (self.credentials.smtp, self.credentials.imap)
|
|
for field in ("username", "password")
|
|
)
|
|
|
|
|
|
class MailServerProfileResponse(BaseModel):
|
|
id: str
|
|
tenant_id: str | None = None
|
|
scope_type: MailProfileScope = "tenant"
|
|
scope_id: str | None = None
|
|
name: str
|
|
slug: str
|
|
description: str | None = None
|
|
is_active: bool
|
|
smtp: dict[str, Any]
|
|
imap: dict[str, Any] | None = None
|
|
credentials: dict[str, Any] = Field(default_factory=dict)
|
|
smtp_password_configured: bool = False
|
|
imap_password_configured: bool = False
|
|
created_at: datetime
|
|
updated_at: datetime
|
|
|
|
|
|
class MailServerProfileListResponse(BaseModel):
|
|
profiles: list[MailServerProfileResponse] = Field(default_factory=list)
|
|
|
|
|
|
class MailSettingsDeltaResponse(BaseModel):
|
|
profiles: list[MailServerProfileResponse] = Field(default_factory=list)
|
|
policy: MailProfilePolicyResponse | None = None
|
|
changed_sections: list[str] = Field(default_factory=list)
|
|
deleted: list[DeltaDeletedItem] = Field(default_factory=list)
|
|
watermark: str | None = None
|
|
has_more: bool = False
|
|
full: bool = False
|
|
|
|
|
|
class MailConnectionTestResponse(BaseModel):
|
|
ok: bool
|
|
protocol: Literal["smtp", "imap"]
|
|
host: str | None = None
|
|
port: int | None = None
|
|
security: str | None = None
|
|
message: str
|
|
details: dict[str, Any] = Field(default_factory=dict)
|
|
|
|
|
|
class MailImapFolderResponse(BaseModel):
|
|
name: str
|
|
flags: list[str] = Field(default_factory=list)
|
|
message_count: int | None = None
|
|
unseen_count: int | None = None
|
|
|
|
|
|
class MailImapFolderListResponse(BaseModel):
|
|
ok: bool
|
|
protocol: Literal["imap"] = "imap"
|
|
host: str | None = None
|
|
port: int | None = None
|
|
security: str | None = None
|
|
message: str
|
|
folders: list[MailImapFolderResponse] = Field(default_factory=list)
|
|
detected_sent_folder: str | None = None
|
|
details: dict[str, Any] = Field(default_factory=dict)
|
|
|
|
class MailMailboxAttachmentResponse(BaseModel):
|
|
filename: str | None = None
|
|
content_type: str
|
|
size_bytes: int
|
|
|
|
|
|
class MailMailboxMessageSummaryResponse(BaseModel):
|
|
uid: str
|
|
folder: str
|
|
subject: str | None = None
|
|
from_header: str | None = None
|
|
to_header: str | None = None
|
|
cc_header: str | None = None
|
|
date: str | None = None
|
|
message_id: str | None = None
|
|
flags: list[str] = Field(default_factory=list)
|
|
size_bytes: int | None = None
|
|
body_preview: str | None = None
|
|
attachment_count: int = 0
|
|
|
|
|
|
class MailMailboxMessageDetailResponse(MailMailboxMessageSummaryResponse):
|
|
body_text: str | None = None
|
|
body_html: str | None = None
|
|
headers: dict[str, str] = Field(default_factory=dict)
|
|
attachments: list[MailMailboxAttachmentResponse] = Field(default_factory=list)
|
|
|
|
|
|
class MailMailboxMessageListResponse(BaseModel):
|
|
profile_id: str
|
|
folder: str
|
|
host: str | None = None
|
|
port: int | None = None
|
|
security: str | None = None
|
|
total_count: int = 0
|
|
offset: int = 0
|
|
limit: int = 50
|
|
cursor: str | None = None
|
|
next_cursor: str | None = None
|
|
cursor_stable: bool = False
|
|
full: bool = False
|
|
messages: list[MailMailboxMessageSummaryResponse] = Field(default_factory=list)
|
|
|
|
|
|
class MailMailboxMessageResponse(BaseModel):
|
|
profile_id: str
|
|
folder: str
|
|
host: str | None = None
|
|
port: int | None = None
|
|
security: str | None = None
|
|
message: MailMailboxMessageDetailResponse
|