austnv/remnawave-minishop
1
0
Fork 0
Code Issues 1 Pull Requests Activity

fix: embed uploaded logo in emails

Browse Source
This commit is contained in:
3252a8
2026-06-03 11:06:19 +03:00
parent 101119911a
commit 2d43033f98
4 changed files with 259 additions and 33 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/bot/services/email_auth_service.py
+76 -23
View File
@@ -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,
*,
backend/bot/services/email_templates.py
+98 -10
View File
@@ -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"""<!DOCTYPE html>
layout_html = f"""<!DOCTYPE html>
<html lang="{html_lang}" xmlns="http://www.w3.org/1999/xhtml">
<head>
<meta charset="utf-8">
@@ -149,6 +227,16 @@ def _layout(
</body>
</html>
""" # 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(
tests/test_email_auth_service.py
+41
View File
@@ -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='<img src="cid:webapp-logo" alt="">',
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"] == "<webapp-logo>"
assert related_images[0].get_content_disposition() == "inline"
tests/test_email_localization.py
+44
View File
@@ -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:]]