140 lines
4.9 KiB
Python
140 lines
4.9 KiB
Python
from __future__ import annotations
|
|
|
|
import urllib.parse
|
|
from typing import Literal
|
|
|
|
from pydantic import Field, model_validator
|
|
|
|
from govoplan_core.mail.config import (
|
|
ImapConfig,
|
|
ImapFolderMappings,
|
|
ImapServerConfig,
|
|
SmtpConfig,
|
|
SmtpServerConfig,
|
|
StrictModel,
|
|
TransportCredentials,
|
|
TransportSecurity,
|
|
normalize_split_transport_credentials,
|
|
)
|
|
|
|
|
|
class Pop3ServerConfig(StrictModel):
|
|
"""Server-only settings for an explicitly enabled legacy POP3 source."""
|
|
|
|
host: str | None = None
|
|
port: int | None = Field(default=None, ge=1, le=65535)
|
|
security: TransportSecurity = TransportSecurity.TLS
|
|
timeout_seconds: int = Field(default=30, ge=1, le=300)
|
|
max_message_bytes: int = Field(default=25 * 1024 * 1024, ge=1_024, le=50 * 1024 * 1024)
|
|
max_batch_bytes: int = Field(default=100 * 1024 * 1024, ge=1_048_576, le=500 * 1024 * 1024)
|
|
preview_body_lines: int = Field(default=20, ge=0, le=100)
|
|
legacy_import_enabled: bool = False
|
|
allow_delete_after_import: bool = False
|
|
|
|
@model_validator(mode="after")
|
|
def apply_default_port(self) -> "Pop3ServerConfig":
|
|
if self.port is None:
|
|
self.port = 995 if self.security == TransportSecurity.TLS else 110
|
|
if self.legacy_import_enabled and not str(self.host or "").strip():
|
|
raise ValueError(
|
|
"POP3 host is required when legacy import is enabled"
|
|
)
|
|
if self.max_batch_bytes < self.max_message_bytes:
|
|
raise ValueError(
|
|
"POP3 batch size limit cannot be lower than the per-message limit"
|
|
)
|
|
if self.allow_delete_after_import and not self.legacy_import_enabled:
|
|
raise ValueError(
|
|
"POP3 delete-after-import cannot be enabled while legacy import is disabled"
|
|
)
|
|
return self
|
|
|
|
|
|
class Pop3Config(Pop3ServerConfig):
|
|
username: str | None = None
|
|
password: str | None = None
|
|
|
|
|
|
class JmapServerConfig(StrictModel):
|
|
"""Server-only settings for an RFC 8620/8621 mailbox endpoint."""
|
|
|
|
session_url: str
|
|
account_id: str | None = Field(default=None, max_length=255)
|
|
auth_scheme: Literal["bearer", "basic"] = "bearer"
|
|
timeout_seconds: int = Field(default=20, ge=1, le=120)
|
|
max_response_bytes: int = Field(
|
|
default=5 * 1024 * 1024,
|
|
ge=64 * 1024,
|
|
le=25 * 1024 * 1024,
|
|
)
|
|
max_body_value_bytes: int = Field(
|
|
default=1 * 1024 * 1024,
|
|
ge=1_024,
|
|
le=5 * 1024 * 1024,
|
|
)
|
|
allowed_api_origins: list[str] = Field(default_factory=list, max_length=10)
|
|
|
|
@model_validator(mode="after")
|
|
def validate_urls(self) -> "JmapServerConfig":
|
|
self.session_url = _absolute_http_url(self.session_url, label="JMAP session URL")
|
|
session_origin = _http_origin(self.session_url)
|
|
origins: list[str] = []
|
|
for value in self.allowed_api_origins:
|
|
normalized = _http_origin(
|
|
_absolute_http_url(value, label="JMAP allowed API origin")
|
|
)
|
|
if normalized != session_origin and normalized not in origins:
|
|
origins.append(normalized)
|
|
self.allowed_api_origins = origins
|
|
if self.account_id is not None:
|
|
self.account_id = self.account_id.strip() or None
|
|
return self
|
|
|
|
|
|
class JmapConfig(JmapServerConfig):
|
|
username: str | None = Field(default=None, max_length=320)
|
|
password: str | None = None
|
|
|
|
@model_validator(mode="after")
|
|
def validate_credentials(self) -> "JmapConfig":
|
|
if self.auth_scheme == "basic" and not (self.username and self.password):
|
|
raise ValueError("JMAP Basic authentication requires username and password")
|
|
if self.auth_scheme == "bearer" and not self.password:
|
|
raise ValueError("JMAP Bearer authentication requires an access token")
|
|
return self
|
|
|
|
|
|
def _absolute_http_url(value: str, *, label: str) -> str:
|
|
parsed = urllib.parse.urlsplit(str(value or "").strip())
|
|
if parsed.scheme not in {"http", "https"} or not parsed.netloc:
|
|
raise ValueError(f"{label} must be an absolute HTTP(S) URL")
|
|
if parsed.username or parsed.password:
|
|
raise ValueError(f"{label} must not include embedded credentials")
|
|
if parsed.fragment:
|
|
raise ValueError(f"{label} must not include a fragment")
|
|
return urllib.parse.urlunsplit(parsed)
|
|
|
|
|
|
def _http_origin(value: str) -> str:
|
|
parsed = urllib.parse.urlsplit(value)
|
|
port = parsed.port or (443 if parsed.scheme == "https" else 80)
|
|
default_port = 443 if parsed.scheme == "https" else 80
|
|
suffix = "" if port == default_port else f":{port}"
|
|
return f"{parsed.scheme.lower()}://{(parsed.hostname or '').lower()}{suffix}"
|
|
|
|
__all__ = [
|
|
"ImapConfig",
|
|
"ImapFolderMappings",
|
|
"ImapServerConfig",
|
|
"JmapConfig",
|
|
"JmapServerConfig",
|
|
"Pop3Config",
|
|
"Pop3ServerConfig",
|
|
"SmtpConfig",
|
|
"SmtpServerConfig",
|
|
"StrictModel",
|
|
"TransportCredentials",
|
|
"TransportSecurity",
|
|
"normalize_split_transport_credentials",
|
|
]
|