add-ideas/govoplan-mail
1
0
Fork 0
Code Issues 9 Pull Requests Actions Packages Projects Releases Wiki Activity

initial commit after split

Browse Source
This commit is contained in:
Albrecht Degering
2026-06-24 01:43:21 +02:00
parent 23213b15e2
commit d922b7701c
47 changed files with 4432 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

0
src/govoplan_mail/backend/sending/__init__.py Normal file
View File

341
src/govoplan_mail/backend/sending/imap.py Normal file
View File

@@ -0,0 +1,341 @@
from __future__ import annotations
import imaplib
import re
import socket
import ssl
import time
from dataclasses import dataclass
from govoplan_mail.backend.config import ImapConfig, TransportSecurity
from govoplan_mail.backend.dev.mock_mailbox import (
MOCK_IMAP_FOLDERS,
consume_fail_next_imap,
is_mock_imap_host,
record_imap_append,
)
class ImapConfigurationError(ValueError):
"""Raised when IMAP settings are incomplete or inconsistent."""
class ImapAppendError(RuntimeError):
"""Raised when APPENDing to Sent fails.
temporary=True means retrying later may help. temporary=False means the
configuration or mailbox choice probably needs user/admin attention.
"""
def __init__(self, message: str, *, temporary: bool | None = None):
super().__init__(message)
self.temporary = temporary
@dataclass(frozen=True, slots=True)
class ImapLoginTestResult:
host: str
port: int
security: str
authenticated: bool
@dataclass(frozen=True, slots=True)
class ImapMailboxInfo:
name: str
flags: list[str]
@dataclass(frozen=True, slots=True)
class ImapFolderListResult:
host: str
port: int
security: str
folders: list[ImapMailboxInfo]
detected_sent_folder: str | None = None
@dataclass(frozen=True, slots=True)
class ImapAppendResult:
host: str
port: int
security: str
folder: str
bytes_appended: int
response: str | None = None
def _require_imap_config(config: ImapConfig) -> tuple[str, int]:
if not config.enabled:
raise ImapConfigurationError("IMAP is disabled")
if not config.host:
raise ImapConfigurationError("IMAP host is required")
if not config.port:
raise ImapConfigurationError("IMAP port is required")
if bool(config.username) != bool(config.password):
raise ImapConfigurationError("IMAP username and password must be provided together, or both omitted")
return config.host, config.port
def _open_imap(config: ImapConfig) -> imaplib.IMAP4:
host, port = _require_imap_config(config)
context = ssl.create_default_context()
try:
if config.security == TransportSecurity.TLS:
client: imaplib.IMAP4 = imaplib.IMAP4_SSL(host=host, port=port, timeout=config.timeout_seconds, ssl_context=context)
else:
client = imaplib.IMAP4(host=host, port=port, timeout=config.timeout_seconds)
if config.security == TransportSecurity.STARTTLS:
typ, data = client.starttls(ssl_context=context)
if typ != "OK":
raise ImapAppendError(f"IMAP STARTTLS failed: {data!r}", temporary=True)
if config.username and config.password:
typ, data = client.login(config.username, config.password)
if typ != "OK":
raise ImapAppendError(f"IMAP login failed: {data!r}", temporary=False)
return client
except Exception:
try:
client.logout() # type: ignore[possibly-undefined]
except Exception:
pass
raise
def _decode_item(item: bytes | str | None) -> str:
if item is None:
return ""
if isinstance(item, bytes):
return item.decode("utf-8", errors="replace")
return item
def _unquote_imap_token(value: str) -> str:
value = value.strip()
if len(value) >= 2 and value[0] == '"' and value[-1] == '"':
value = value[1:-1]
value = value.replace('\\"', '"').replace('\\\\', '\\')
return value
def _extract_mailbox_name(list_response_line: bytes | str) -> tuple[str, set[str]] | None:
r"""Best-effort parser for IMAP LIST response lines.
RFC 3501 LIST responses contain attributes, hierarchy delimiter, then mailbox
name. Some servers quote both delimiter and mailbox::
(\HasNoChildren \Sent) "/" "Sent"
Others quote only the delimiter and leave the mailbox as an atom::
(\HasNoChildren \Sent) "/" Sent
The parser must therefore parse the delimiter token separately instead of
blindly taking the last quoted value.
"""
line = _decode_item(list_response_line).strip()
match = re.match(
r'^\((?P<flags>[^)]*)\)\s+'
r'(?P<delimiter>"(?:[^"\\]|\\.)*"|NIL|[^\s]+)\s+'
r'(?P<mailbox>.+?)\s*$',
line,
re.IGNORECASE,
)
if match:
flags = {part.lower() for part in match.group("flags").split()}
mailbox = _unquote_imap_token(match.group("mailbox"))
if mailbox:
return mailbox, flags
return None
# Fallback for non-standard server lines: prefer the final token.
parts = line.split(maxsplit=2)
if len(parts) >= 3:
flags_text = parts[0].strip("()")
flags = {part.lower() for part in flags_text.split()}
mailbox = _unquote_imap_token(parts[2])
if mailbox:
return mailbox, flags
return None
def _detect_sent_folder(parsed: list[tuple[str, set[str]]]) -> str | None:
for name, flags in parsed:
if "\\sent" in flags or "\\sentmail" in flags:
return name
common_names = [
"Sent",
"Sent Items",
"Sent Messages",
"Gesendet",
"Gesendete Elemente",
"INBOX.Sent",
"INBOX/Sent",
]
names = {name.lower(): name for name, _ in parsed}
for candidate in common_names:
if candidate.lower() in names:
return names[candidate.lower()]
return None
def discover_sent_folder(client: imaplib.IMAP4) -> str | None:
typ, data = client.list()
if typ != "OK" or not data:
return None
parsed: list[tuple[str, set[str]]] = []
for item in data:
extracted = _extract_mailbox_name(item)
if extracted:
parsed.append(extracted)
return _detect_sent_folder(parsed)
def _effective_sent_folder(*, config: ImapConfig, requested_folder: str | None, client: imaplib.IMAP4) -> str:
if requested_folder and requested_folder != "auto":
return requested_folder
if config.sent_folder and config.sent_folder != "auto":
return config.sent_folder
discovered = discover_sent_folder(client)
if discovered:
return discovered
raise ImapConfigurationError("Could not discover Sent folder; configure delivery.imap_append_sent.folder or server.imap.sent_folder")
def test_imap_login(*, imap_config: ImapConfig) -> ImapLoginTestResult:
"""Open an IMAP connection and authenticate if credentials are configured.
This is a side-effect-free connection test for the WebUI. It does not select
a mailbox and does not append any message.
"""
host, port = _require_imap_config(imap_config)
if is_mock_imap_host(imap_config.host):
return ImapLoginTestResult(
host=host,
port=port,
security=imap_config.security.value,
authenticated=bool(imap_config.username and imap_config.password),
)
client = _open_imap(imap_config)
try:
return ImapLoginTestResult(
host=host,
port=port,
security=imap_config.security.value,
authenticated=bool(imap_config.username and imap_config.password),
)
finally:
try:
client.logout()
except Exception:
pass
def list_imap_folders(*, imap_config: ImapConfig) -> ImapFolderListResult:
"""Return folders visible through IMAP LIST and the best sent-folder guess."""
host, port = _require_imap_config(imap_config)
if is_mock_imap_host(imap_config.host):
folders = [ImapMailboxInfo(name=str(item["name"]), flags=list(item.get("flags") or [])) for item in MOCK_IMAP_FOLDERS]
return ImapFolderListResult(
host=host,
port=port,
security=imap_config.security.value,
folders=folders,
detected_sent_folder="Sent",
)
client = _open_imap(imap_config)
try:
typ, data = client.list()
if typ != "OK":
raise ImapAppendError(f"IMAP folder listing failed: {data!r}", temporary=True)
parsed: list[tuple[str, set[str]]] = []
folders: list[ImapMailboxInfo] = []
for item in data or []:
extracted = _extract_mailbox_name(item)
if not extracted:
continue
name, flags = extracted
parsed.append((name, flags))
folders.append(ImapMailboxInfo(name=name, flags=sorted(flags)))
return ImapFolderListResult(
host=host,
port=port,
security=imap_config.security.value,
folders=folders,
detected_sent_folder=_detect_sent_folder(parsed),
)
finally:
try:
client.logout()
except Exception:
pass
def append_message_to_sent(
message_bytes: bytes,
*,
imap_config: ImapConfig,
folder: str | None = None,
) -> ImapAppendResult:
"""Append a sent MIME message to the configured IMAP Sent folder.
The SMTP send remains authoritative. APPEND is a separate best-effort step
and should not be used to decide whether an email was sent.
"""
host, port = _require_imap_config(imap_config)
if is_mock_imap_host(imap_config.host):
if consume_fail_next_imap():
raise ImapAppendError("Mock IMAP configured to fail the next append", temporary=False)
target_folder = folder or (imap_config.sent_folder if imap_config.sent_folder and imap_config.sent_folder != "auto" else "Sent")
record = record_imap_append(message_bytes, folder=target_folder, imap_host=imap_config.host)
return ImapAppendResult(
host=host,
port=port,
security=imap_config.security.value,
folder=target_folder,
bytes_appended=len(message_bytes),
response=f"mock append stored as {record.id}",
)
client: imaplib.IMAP4 | None = None
try:
client = _open_imap(imap_config)
target_folder = _effective_sent_folder(config=imap_config, requested_folder=folder, client=client)
internal_date = imaplib.Time2Internaldate(time.time())
typ, data = client.append(target_folder, "\\Seen", internal_date, message_bytes)
if typ != "OK":
raise ImapAppendError(f"IMAP APPEND failed for folder {target_folder!r}: {data!r}", temporary=False)
response = "; ".join(_decode_item(item) for item in (data or [])) or None
return ImapAppendResult(
host=host,
port=port,
security=imap_config.security.value,
folder=target_folder,
bytes_appended=len(message_bytes),
response=response,
)
except ImapAppendError:
raise
except (OSError, socket.timeout, imaplib.IMAP4.abort) as exc:
raise ImapAppendError(f"IMAP append failed: {exc}", temporary=True) from exc
except imaplib.IMAP4.error as exc:
raise ImapAppendError(f"IMAP append failed: {exc}", temporary=False) from exc
finally:
if client is not None:
try:
client.logout()
except Exception:
pass

57
src/govoplan_mail/backend/sending/rate_limit.py Normal file
View File

@@ -0,0 +1,57 @@
from __future__ import annotations
import time
from dataclasses import dataclass
from redis import Redis
from redis.exceptions import RedisError
from govoplan_mail.backend.runtime import settings
@dataclass(frozen=True, slots=True)
class RateLimitDecision:
key: str
messages_per_minute: int
gap_seconds: float
waited_seconds: float
def _redis_client() -> Redis:
return Redis.from_url(settings.redis_url, decode_responses=True)
def wait_for_rate_limit(*, key: str, messages_per_minute: int, enabled: bool = True) -> RateLimitDecision:
"""Throttle sends across worker processes using Redis when available.
The implementation stores the next allowed send timestamp per key. A Redis
lock keeps multiple Celery processes from reading/updating the timestamp at
the same time. If Redis is unavailable, it falls back to no distributed wait;
the per-container Celery concurrency still protects local development.
"""
messages_per_minute = max(1, int(messages_per_minute or 1))
gap = 60.0 / messages_per_minute
if not enabled:
return RateLimitDecision(key=key, messages_per_minute=messages_per_minute, gap_seconds=gap, waited_seconds=0.0)
redis_key = f"multimailer:ratelimit:{key}:next_allowed"
lock_key = f"multimailer:ratelimit:{key}:lock"
waited = 0.0
try:
client = _redis_client()
with client.lock(lock_key, timeout=30, blocking_timeout=30):
now = time.time()
raw_next = client.get(redis_key)
next_allowed = float(raw_next) if raw_next else now
if next_allowed > now:
waited = next_allowed - now
time.sleep(waited)
now = time.time()
client.set(redis_key, now + gap, ex=max(60, int(gap * 10)))
except (RedisError, TimeoutError, ValueError):
# Development fallback: do not fail sending because Redis is absent.
waited = 0.0
return RateLimitDecision(key=key, messages_per_minute=messages_per_minute, gap_seconds=gap, waited_seconds=waited)

322
src/govoplan_mail/backend/sending/smtp.py Normal file
View File

@@ -0,0 +1,322 @@
from __future__ import annotations
import copy
import smtplib
import ssl
from dataclasses import dataclass
from email.message import EmailMessage
from email.utils import formataddr
from govoplan_mail.backend.config import SmtpConfig, TransportSecurity
from govoplan_mail.backend.dev.mock_mailbox import (
consume_fail_next_smtp,
get_failures,
is_mock_smtp_host,
record_smtp_delivery,
)
class SmtpConfigurationError(ValueError):
"""Raised when SMTP settings are incomplete or inconsistent."""
class SmtpSendError(RuntimeError):
"""Raised when an SMTP send attempt fails.
``temporary`` means an explicit response allows a later retry. An
``outcome_unknown`` failure happened after SMTP transmission may have
started, so automatic retry is intentionally forbidden.
"""
def __init__(self, message: str, *, temporary: bool = False, outcome_unknown: bool = False):
super().__init__(message)
self.temporary = temporary
self.outcome_unknown = outcome_unknown
@dataclass(frozen=True, slots=True)
class SmtpLoginTestResult:
host: str
port: int
security: str
authenticated: bool
@dataclass(frozen=True, slots=True)
class SmtpSendResult:
host: str
port: int
security: str
envelope_from: str
envelope_recipients: list[str]
refused_recipients: dict[str, tuple[int, bytes | str]]
@property
def accepted_count(self) -> int:
return len(self.envelope_recipients) - len(self.refused_recipients)
def _require_smtp_config(config: SmtpConfig) -> tuple[str, int]:
if not config.host:
raise SmtpConfigurationError("SMTP host is required")
if not config.port:
raise SmtpConfigurationError("SMTP port is required")
if bool(config.username) != bool(config.password):
raise SmtpConfigurationError("SMTP username and password must be provided together, or both omitted")
return config.host, config.port
def _open_smtp(config: SmtpConfig) -> smtplib.SMTP:
host, port = _require_smtp_config(config)
context = ssl.create_default_context()
try:
if config.security == TransportSecurity.TLS:
smtp: smtplib.SMTP = smtplib.SMTP_SSL(host=host, port=port, timeout=config.timeout_seconds, context=context)
smtp.ehlo()
else:
smtp = smtplib.SMTP(host=host, port=port, timeout=config.timeout_seconds)
smtp.ehlo()
if config.security == TransportSecurity.STARTTLS:
smtp.starttls(context=context)
smtp.ehlo()
if config.username and config.password:
smtp.login(config.username, config.password)
return smtp
except Exception:
# If construction/login fails after a socket was created, smtplib usually closes
# on GC, but explicit cleanup is safer when the variable exists.
try:
smtp.quit() # type: ignore[possibly-undefined]
except Exception:
pass
raise
def _decode_refused(refused: dict[str, tuple[int, bytes]]) -> dict[str, tuple[int, bytes | str]]:
normalized: dict[str, tuple[int, bytes | str]] = {}
for recipient, (code, response) in refused.items():
try:
normalized[recipient] = (code, response.decode("utf-8", errors="replace"))
except AttributeError:
normalized[recipient] = (code, response)
return normalized
def test_smtp_login(*, smtp_config: SmtpConfig) -> SmtpLoginTestResult:
"""Open an SMTP connection and authenticate if credentials are configured.
This is intentionally side-effect free: it does not send a message and it
never receives envelope or recipient data. It is used by the WebUI to check
whether the configured transport can be reached before a campaign is built
or queued.
"""
host, port = _require_smtp_config(smtp_config)
if is_mock_smtp_host(smtp_config.host):
host, port = _require_smtp_config(smtp_config)
return SmtpLoginTestResult(
host=host,
port=port,
security=smtp_config.security.value,
authenticated=bool(smtp_config.username and smtp_config.password),
)
smtp = _open_smtp(smtp_config)
try:
return SmtpLoginTestResult(
host=host,
port=port,
security=smtp_config.security.value,
authenticated=bool(smtp_config.username and smtp_config.password),
)
finally:
try:
smtp.quit()
except Exception:
try:
smtp.close()
except Exception:
pass
def prepare_test_message(
message: EmailMessage,
*,
test_recipient: str,
test_recipient_name: str | None = None,
) -> EmailMessage:
"""Return a safe copy of a generated campaign message for test delivery.
The original recipient headers are removed so a test send cannot accidentally
leak the real To/Cc list or deliver to the real recipients. The envelope
recipient must also be supplied separately to send_email_message().
"""
test_message = copy.deepcopy(message)
for header in ["To", "Cc", "Bcc"]:
if header in test_message:
del test_message[header]
# Replace potential previous marker headers if the user test-sends an EML twice.
for header in ["X-MultiMailer-Test-Send"]:
if header in test_message:
del test_message[header]
test_message["To"] = formataddr((test_recipient_name or test_recipient, test_recipient))
test_message["X-MultiMailer-Test-Send"] = "true"
return test_message
def _send_smtp_payload(
message: EmailMessage | bytes,
*,
smtp_config: SmtpConfig,
envelope_from: str,
envelope_recipients: list[str],
) -> SmtpSendResult:
host, port = _require_smtp_config(smtp_config)
if not envelope_from:
raise SmtpConfigurationError("SMTP envelope sender is required")
if not envelope_recipients:
raise SmtpConfigurationError("at least one SMTP envelope recipient is required")
if is_mock_smtp_host(smtp_config.host):
if consume_fail_next_smtp():
raise SmtpSendError("Mock SMTP configured to fail the next send")
failures = get_failures()
reject_text = str(failures.get("smtp_reject_recipients_containing") or "").strip().lower()
refused: dict[str, tuple[int, bytes]] = {}
accepted = list(envelope_recipients)
if reject_text:
refused = {
recipient: (550, b"mock recipient rejected")
for recipient in envelope_recipients
if reject_text in recipient.lower()
}
accepted = [recipient for recipient in envelope_recipients if recipient not in refused]
if not accepted:
raise SmtpSendError(f"all mock SMTP recipients were refused: {_decode_refused(refused)}")
record_smtp_delivery(
message,
envelope_from=envelope_from,
envelope_recipients=accepted,
smtp_host=smtp_config.host,
)
return SmtpSendResult(
host=host,
port=port,
security=smtp_config.security.value,
envelope_from=envelope_from,
envelope_recipients=list(envelope_recipients),
refused_recipients=_decode_refused(refused),
)
try:
smtp = _open_smtp(smtp_config)
except smtplib.SMTPAuthenticationError as exc:
raise SmtpSendError(
f"SMTP authentication failed: {exc.smtp_code} {exc.smtp_error!r}",
temporary=False,
) from exc
except smtplib.SMTPResponseException as exc:
raise SmtpSendError(
f"SMTP connection error: {exc.smtp_code} {exc.smtp_error!r}",
temporary=400 <= int(exc.smtp_code) < 500,
) from exc
except (OSError, smtplib.SMTPException) as exc:
# No message transmission has begun yet; a later explicit retry is safe.
raise SmtpSendError(f"SMTP connection failed: {exc}", temporary=True) from exc
try:
if isinstance(message, bytes):
refused = smtp.sendmail(envelope_from, envelope_recipients, message)
else:
refused = smtp.send_message(
message,
from_addr=envelope_from,
to_addrs=envelope_recipients,
)
except smtplib.SMTPRecipientsRefused as exc:
raise SmtpSendError(
f"all SMTP recipients were refused: {_decode_refused(exc.recipients)}",
temporary=False,
) from exc
except smtplib.SMTPSenderRefused as exc:
raise SmtpSendError(
f"SMTP sender was refused: {exc.smtp_code} {exc.smtp_error!r}",
temporary=400 <= int(exc.smtp_code) < 500,
) from exc
except smtplib.SMTPResponseException as exc:
# An explicit SMTP response means the server rejected the transaction;
# retryability follows the response class.
raise SmtpSendError(
f"SMTP error: {exc.smtp_code} {exc.smtp_error!r}",
temporary=400 <= int(exc.smtp_code) < 500,
) from exc
except (OSError, smtplib.SMTPServerDisconnected, smtplib.SMTPException) as exc:
# A connection loss after DATA began can happen after the server accepted
# the message but before the client received the final response.
raise SmtpSendError(
f"SMTP outcome is unknown after transmission started: {exc}",
outcome_unknown=True,
) from exc
finally:
try:
smtp.quit()
except Exception:
try:
smtp.close()
except Exception:
pass
return SmtpSendResult(
host=host,
port=port,
security=smtp_config.security.value,
envelope_from=envelope_from,
envelope_recipients=list(envelope_recipients),
refused_recipients=_decode_refused(refused),
)
def send_email_bytes(
message_bytes: bytes,
*,
smtp_config: SmtpConfig,
envelope_from: str,
envelope_recipients: list[str],
) -> SmtpSendResult:
"""Send exact RFC 5322 bytes through SMTP without reserializing the message."""
return _send_smtp_payload(
message_bytes,
smtp_config=smtp_config,
envelope_from=envelope_from,
envelope_recipients=envelope_recipients,
)
def send_email_message(
message: EmailMessage,
*,
smtp_config: SmtpConfig,
envelope_from: str,
envelope_recipients: list[str],
) -> SmtpSendResult:
"""Send an EmailMessage through SMTP.
This low-level function deliberately receives explicit envelope sender and
recipients. Headers and SMTP envelope are related but not identical; Bcc and
future bounce-address handling depend on keeping them separate. Campaign
delivery uses send_email_bytes() so the generated EML is not reserialized.
"""
return _send_smtp_payload(
message,
smtp_config=smtp_config,
envelope_from=envelope_from,
envelope_recipients=envelope_recipients,
)