diff --git a/backend/bot/services/email_auth_service.py b/backend/bot/services/email_auth_service.py index 9391932..034f780 100644 --- a/backend/bot/services/email_auth_service.py +++ b/backend/bot/services/email_auth_service.py @@ -10,13 +10,13 @@ from dataclasses import dataclass from datetime import datetime, timedelta, timezone from email.message import EmailMessage from email.utils import formataddr -from typing import Optional +from typing import Optional, Sequence from sqlalchemy import select, update from sqlalchemy.ext.asyncio import AsyncSession from bot.middlewares.i18n import JsonI18n -from bot.services.email_templates import EmailContent, render_login_code +from bot.services.email_templates import EmailContent, EmailInlineImage, render_login_code from bot.services.message_audit import log_user_message_delivery from config.settings import Settings from db.dal import security_dal, user_dal @@ -453,6 +453,7 @@ class EmailAuthService: subject: str, body: str, html_body: Optional[str] = None, + inline_images: Sequence[EmailInlineImage] = (), ) -> None: await asyncio.to_thread( self._send_custom_email_sync, @@ -460,6 +461,7 @@ class EmailAuthService: subject=subject, body=body, html_body=html_body, + inline_images=inline_images, ) async def send_rendered_email( @@ -473,6 +475,7 @@ class EmailAuthService: subject=content.subject, body=content.text, html_body=content.html, + inline_images=content.inline_images, ) def _send_code_email_sync( @@ -493,17 +496,13 @@ class EmailAuthService: i18n=self.i18n, ) - message = EmailMessage() - message["Subject"] = content.subject - message["From"] = formataddr( - ( - self.settings.SMTP_FROM_NAME or self.settings.WEBAPP_TITLE, - self.settings.SMTP_FROM_EMAIL or "", - ) + message = self._build_email_message( + email=email, + subject=content.subject, + body=content.text, + html_body=content.html, + inline_images=content.inline_images, ) - message["To"] = email - message.set_content(content.text) - message.add_alternative(content.html, subtype="html") context = ssl.create_default_context() smtp_host = self.settings.SMTP_HOST @@ -554,19 +553,15 @@ class EmailAuthService: subject: str, body: str, html_body: Optional[str] = None, + inline_images: Sequence[EmailInlineImage] = (), ) -> None: - message = EmailMessage() - message["Subject"] = subject - message["From"] = formataddr( - ( - self.settings.SMTP_FROM_NAME or self.settings.WEBAPP_TITLE, - self.settings.SMTP_FROM_EMAIL or "", - ) + message = self._build_email_message( + email=email, + subject=subject, + body=body, + html_body=html_body, + inline_images=inline_images, ) - message["To"] = email - message.set_content(body) - if html_body: - message.add_alternative(html_body, subtype="html") context = ssl.create_default_context() smtp_host = self.settings.SMTP_HOST @@ -610,6 +605,64 @@ class EmailAuthService: if last_error: raise last_error + def _build_email_message( + self, + *, + email: str, + subject: str, + body: str, + html_body: Optional[str] = None, + inline_images: Sequence[EmailInlineImage] = (), + ) -> EmailMessage: + message = EmailMessage() + message["Subject"] = subject + message["From"] = formataddr( + ( + self.settings.SMTP_FROM_NAME or self.settings.WEBAPP_TITLE, + self.settings.SMTP_FROM_EMAIL or "", + ) + ) + message["To"] = email + message.set_content(body) + if html_body: + message.add_alternative(html_body, subtype="html") + self._attach_inline_images(message, inline_images) + return message + + @staticmethod + def _attach_inline_images( + message: EmailMessage, + inline_images: Sequence[EmailInlineImage], + ) -> None: + if not inline_images: + return + html_part = message.get_body(("html",)) + if html_part is None: + return + + for image in inline_images: + content_type = (image.content_type or "").split(";", 1)[0].strip().lower() + if "/" not in content_type: + continue + maintype, subtype = content_type.split("/", 1) + if maintype != "image" or not subtype: + continue + + body = bytes(image.data or b"") + content_id = (image.content_id or "").strip() + if not body or not content_id: + continue + + cid_header = content_id + if not (cid_header.startswith("<") and cid_header.endswith(">")): + cid_header = f"<{cid_header}>" + html_part.add_related( + body, + maintype=maintype, + subtype=subtype, + cid=cid_header, + ) + def _send_message_via_smtp( self, *, diff --git a/backend/bot/services/email_templates.py b/backend/bot/services/email_templates.py index 0b09c1f..57da1eb 100644 --- a/backend/bot/services/email_templates.py +++ b/backend/bot/services/email_templates.py @@ -12,6 +12,7 @@ from __future__ import annotations import html import re from dataclasses import dataclass +from pathlib import Path from typing import TYPE_CHECKING, Optional, Sequence, Tuple from urllib.parse import urlsplit @@ -27,6 +28,27 @@ _TEXT_MUTED = "#9aa3b2" _TEXT_DIM = "#5d6573" _DEFAULT_ACCENT = "#00fe7a" _HEX_RE = re.compile(r"^#(?:[0-9a-fA-F]{3}|[0-9a-fA-F]{6})$") +_EMAIL_LOGO_CONTENT_ID = "webapp-logo" +_WEBAPP_UPLOADED_LOGO_PATH = "/webapp-uploaded-logo" +_WEBAPP_UPLOADED_LOGO_DIR = Path(__file__).resolve().parents[3] / "data" / "webapp-logo" / "uploads" +_WEBAPP_LOGO_MAX_BYTES = 2 * 1024 * 1024 +_UPLOADED_LOGO_RE = re.compile(r"logo-[0-9a-f]{16}\.(?:gif|ico|jpe?g|png|svg|webp)") +_LOGO_CONTENT_TYPES = { + ".gif": "image/gif", + ".ico": "image/x-icon", + ".jpg": "image/jpeg", + ".jpeg": "image/jpeg", + ".png": "image/png", + ".svg": "image/svg+xml", + ".webp": "image/webp", +} + + +@dataclass(frozen=True) +class EmailInlineImage: + content_id: str + content_type: str + data: bytes @dataclass(frozen=True) @@ -34,6 +56,13 @@ class EmailContent: subject: str text: str html: str + inline_images: Tuple[EmailInlineImage, ...] = () + + +@dataclass(frozen=True) +class _EmailLayout: + html: str + inline_images: Tuple[EmailInlineImage, ...] = () def _safe_color(value: Optional[str]) -> str: @@ -57,6 +86,55 @@ def _public_logo_url(settings: Settings) -> Optional[str]: return raw +def _uploaded_logo_filename(url: str) -> Optional[str]: + parsed = urlsplit(str(url or "")) + path = parsed.path if parsed.scheme or parsed.netloc else str(url or "") + prefix = f"{_WEBAPP_UPLOADED_LOGO_PATH}/" + if not path.startswith(prefix): + return None + filename = path.removeprefix(prefix) + return filename if _UPLOADED_LOGO_RE.fullmatch(filename) else None + + +def _inline_uploaded_logo(settings: Settings) -> Optional[EmailInlineImage]: + filename = _uploaded_logo_filename((settings.WEBAPP_LOGO_URL or "").strip()) + if not filename: + return None + + content_type = _LOGO_CONTENT_TYPES.get(Path(filename).suffix.lower()) + if not content_type: + return None + + try: + uploads_dir = _WEBAPP_UPLOADED_LOGO_DIR.resolve() + logo_path = (uploads_dir / filename).resolve() + logo_path.relative_to(uploads_dir) + body = logo_path.read_bytes() + except (OSError, ValueError): + return None + + if not body or len(body) > _WEBAPP_LOGO_MAX_BYTES: + return None + + return EmailInlineImage( + content_id=_EMAIL_LOGO_CONTENT_ID, + content_type=content_type, + data=body, + ) + + +def _email_logo(settings: Settings) -> Tuple[Optional[str], Tuple[EmailInlineImage, ...]]: + inline_logo = _inline_uploaded_logo(settings) + if inline_logo: + return f"cid:{inline_logo.content_id}", (inline_logo,) + + public_url = _public_logo_url(settings) + if public_url: + return public_url, () + + return None, () + + def _brand_title(settings: Settings) -> str: title = (settings.WEBAPP_TITLE or "").strip() return title or "Subscription" @@ -96,10 +174,10 @@ def _layout( intro_html: str, body_html: str, footer_html: str, -) -> str: +) -> _EmailLayout: accent = _safe_color(settings.WEBAPP_PRIMARY_COLOR) brand_title = html.escape(_brand_title(settings)) - logo_url = _public_logo_url(settings) + logo_url, inline_images = _email_logo(settings) html_lang = html.escape((language_code or "en").replace("_", "-"), quote=True) logo_block = "" if logo_url: @@ -109,7 +187,7 @@ def _layout( f'border-radius:16px;">' ) - return f""" + layout_html = f""" @@ -149,6 +227,16 @@ def _layout( """ # noqa: E501 + return _EmailLayout(html=layout_html, inline_images=inline_images) + + +def _email_content(*, subject: str, text: str, layout: _EmailLayout) -> EmailContent: + return EmailContent( + subject=subject, + text=text, + html=layout.html, + inline_images=layout.inline_images, + ) def _info_rows_html(rows: Sequence[Tuple[str, str]]) -> str: @@ -317,7 +405,7 @@ def render_login_code( body_html=body_html, footer_html=footer, ) - return EmailContent(subject=subject, text="\n".join(text_lines), html=rendered) + return _email_content(subject=subject, text="\n".join(text_lines), layout=rendered) def render_account_merged( @@ -370,7 +458,7 @@ def render_account_merged( body_html=body_html, footer_html=footer, ) - return EmailContent(subject=subject, text=text, html=rendered) + return _email_content(subject=subject, text=text, layout=rendered) def render_payment_success( @@ -504,7 +592,7 @@ def render_payment_success( body_html="".join(body_parts), footer_html=footer, ) - return EmailContent(subject=subject, text="\n".join(text_lines), html=rendered) + return _email_content(subject=subject, text="\n".join(text_lines), layout=rendered) def render_user_notification( @@ -565,7 +653,7 @@ def render_user_notification( ), ] ) - return EmailContent(subject=final_subject, text="\n".join(text_lines), html=rendered) + return _email_content(subject=final_subject, text="\n".join(text_lines), layout=rendered) def render_subscription_expiring( @@ -629,7 +717,7 @@ def render_subscription_expiring( body_html="".join(body_parts), footer_html=footer, ) - return EmailContent(subject=subject, text="\n".join(text_lines), html=rendered) + return _email_content(subject=subject, text="\n".join(text_lines), layout=rendered) def _subscription_lifecycle_title( @@ -731,7 +819,7 @@ def render_subscription_lifecycle_notification( ), ] ) - return EmailContent(subject=subject, text="\n".join(text_lines), html=rendered) + return _email_content(subject=subject, text="\n".join(text_lines), layout=rendered) def _support_email( @@ -783,7 +871,7 @@ def _support_email( ] if safe_url: text_lines.extend(["", safe_url]) - return EmailContent(subject=subject, text="\n".join(text_lines), html=rendered) + return _email_content(subject=subject, text="\n".join(text_lines), layout=rendered) def render_support_new_ticket_admin( diff --git a/tests/test_email_auth_service.py b/tests/test_email_auth_service.py new file mode 100644 index 0000000..5668023 --- /dev/null +++ b/tests/test_email_auth_service.py @@ -0,0 +1,41 @@ +from types import SimpleNamespace + +from bot.services.email_auth_service import EmailAuthService +from bot.services.email_templates import EmailInlineImage + + +def _settings(): + return SimpleNamespace( + SMTP_FROM_NAME="Mini Shop", + SMTP_FROM_EMAIL="noreply@example.com", + WEBAPP_TITLE="Mini Shop", + ) + + +def test_build_email_message_attaches_inline_images_to_html_part(): + service = EmailAuthService(_settings()) + + message = service._build_email_message( + email="user@example.com", + subject="Login code", + body="Your code: 123456", + html_body='', + inline_images=( + EmailInlineImage( + content_id="webapp-logo", + content_type="image/png", + data=b"\x89PNG\r\n\x1a\nlogo", + ), + ), + ) + + related_part = message.get_body(("related",)) + assert related_part is not None + html_part = related_part.get_body(("html",)) + assert html_part is not None + related_images = [part for part in related_part.iter_attachments()] + + assert len(related_images) == 1 + assert related_images[0].get_content_type() == "image/png" + assert related_images[0]["Content-ID"] == "" + assert related_images[0].get_content_disposition() == "inline" diff --git a/tests/test_email_localization.py b/tests/test_email_localization.py index 4ba7838..68a2c95 100644 --- a/tests/test_email_localization.py +++ b/tests/test_email_localization.py @@ -6,6 +6,7 @@ from pathlib import Path from types import SimpleNamespace from bot.middlewares.i18n import JsonI18n +from bot.services import email_templates as email_templates_module from bot.services.email_templates import ( EmailContent, render_account_merged, @@ -255,6 +256,49 @@ def test_all_email_template_variants_render_without_raw_locale_keys(): _assert_content_is_localized(content, language) +def test_uploaded_webapp_logo_is_embedded_inline(tmp_path, monkeypatch): + uploads_dir = tmp_path / "uploads" + uploads_dir.mkdir() + filename = "logo-1111111111111111.png" + logo_body = b"\x89PNG\r\n\x1a\nlogo" + (uploads_dir / filename).write_bytes(logo_body) + monkeypatch.setattr(email_templates_module, "_WEBAPP_UPLOADED_LOGO_DIR", uploads_dir) + + settings = _settings() + settings.WEBAPP_LOGO_URL = f"/webapp-uploaded-logo/{filename}" + + content = render_login_code( + settings, + code="123456", + language_code="en", + purpose="login", + i18n=_i18n("en"), + ) + + assert 'src="cid:webapp-logo"' in content.html + assert len(content.inline_images) == 1 + inline_logo = content.inline_images[0] + assert inline_logo.content_id == "webapp-logo" + assert inline_logo.content_type == "image/png" + assert inline_logo.data == logo_body + + +def test_public_https_webapp_logo_remains_external(): + settings = _settings() + settings.WEBAPP_LOGO_URL = "https://cdn.example.com/logo.png" + + content = render_login_code( + settings, + code="123456", + language_code="en", + purpose="login", + i18n=_i18n("en"), + ) + + assert 'src="https://cdn.example.com/logo.png"' in content.html + assert content.inline_images == () + + def test_support_email_templates_use_russian_copy_for_russian_recipients(): subjects = [content.subject for content in _all_rendered_email_variants("ru")[-4:]]