328 lines
8.9 KiB
Python
328 lines
8.9 KiB
Python
# `email_task.py` generate from `email_task.ipynb` by `build.ipynb`
|
||
#ignore
|
||
import os
|
||
|
||
def load_env(file_path):
|
||
"""Загружает переменные окружения из файла .env с поддержкой булевых значений."""
|
||
with open(file_path) as f:
|
||
for line in f:
|
||
line = line.strip()
|
||
if not line or line.startswith('#'):
|
||
continue
|
||
|
||
try:
|
||
key, value = line.split('=', 1)
|
||
key = key.strip()
|
||
value = value.strip('"\' ')
|
||
# Преобразование строковых значений в булевые
|
||
if value.lower() in ['true', '1', 'yes']:
|
||
value = True
|
||
elif value.lower() in ['false', '0', 'no']:
|
||
value = False
|
||
|
||
# Проверка, что ключ не пустой
|
||
if key:
|
||
os.environ[key] = str(value) # Храним как строку
|
||
except ValueError:
|
||
print(f"Ошибка: неверный формат в строке '{line}'")
|
||
|
||
try:
|
||
load_env('env')
|
||
except:
|
||
load_env('env_example')
|
||
|
||
import smtplib
|
||
import os
|
||
from email.mime.application import MIMEApplication
|
||
from email.mime.multipart import MIMEMultipart
|
||
from email.mime.text import MIMEText
|
||
from email.utils import formatdate, make_msgid
|
||
from typing import Optional
|
||
import gnupg
|
||
from celery import Celery
|
||
from config import *
|
||
from db import *
|
||
from templates import templates
|
||
import logging
|
||
import asyncio
|
||
|
||
|
||
def build_unsubscribe_url(email: str, token: str, base_url: str) -> str:
|
||
return f"{base_url}/api/v1/email/unsubscribe?token={token}&email={email}"
|
||
|
||
|
||
GPG_HOME = os.getenv("GPG_HOME", "/var/lib/app/.gnupg")
|
||
GPG_BINARY = os.getenv("GPG_BINARY", "gpg")
|
||
|
||
gpg = gnupg.GPG(
|
||
gnupghome=GPG_HOME,
|
||
gpgbinary=GPG_BINARY,
|
||
)
|
||
|
||
|
||
def build_unsubscribe_url(
|
||
to_email: str,
|
||
user_id: Optional[int],
|
||
domain: str,
|
||
) -> str:
|
||
"""
|
||
Replace this implementation with your real unsubscribe URL builder.
|
||
Avoid putting a raw email address in the URL in production.
|
||
"""
|
||
from urllib.parse import quote
|
||
|
||
return f"https://{domain}/unsubscribe?email={quote(to_email)}"
|
||
|
||
|
||
def import_public_key(public_key: str) -> str:
|
||
"""
|
||
Import an ASCII-armored public key and return its fingerprint.
|
||
|
||
The database value should contain the complete public key, for example:
|
||
|
||
-----BEGIN PGP PUBLIC KEY BLOCK-----
|
||
...
|
||
-----END PGP PUBLIC KEY BLOCK-----
|
||
"""
|
||
result = gpg.import_keys(public_key)
|
||
|
||
if not result.fingerprints:
|
||
raise RuntimeError(
|
||
f"Unable to import the recipient's public key: {result}"
|
||
)
|
||
|
||
# Use the imported fingerprint rather than the email address. The email
|
||
# address in the key's UID may not exactly match the recipient address.
|
||
return result.fingerprints[0]
|
||
|
||
|
||
def encrypt_mime_message(
|
||
message,
|
||
recipient_fingerprint: str,
|
||
):
|
||
"""
|
||
Encrypt an already-built MIME message and return a PGP/MIME message.
|
||
"""
|
||
|
||
# Serialize the original MIME message before encryption.
|
||
plaintext = message.as_bytes()
|
||
|
||
# Binary OpenPGP data is appropriate for application/octet-stream.
|
||
encrypted = gpg.encrypt(
|
||
plaintext,
|
||
recipients=[recipient_fingerprint],
|
||
armor=False,
|
||
always_trust=False,
|
||
)
|
||
|
||
if not encrypted.ok:
|
||
raise RuntimeError(
|
||
f"GPG encryption failed: {encrypted.status}; "
|
||
f"stderr: {encrypted.stderr}"
|
||
)
|
||
|
||
encrypted_bytes = encrypted.data
|
||
|
||
if not encrypted_bytes:
|
||
raise RuntimeError("GPG returned an empty encrypted message")
|
||
|
||
encrypted_message = MIMEMultipart(
|
||
"encrypted",
|
||
protocol="application/pgp-encrypted",
|
||
)
|
||
|
||
# Copy normal mail headers. Content-Type and MIME-Version are generated
|
||
# automatically for the new multipart/encrypted message.
|
||
for header in (
|
||
"From",
|
||
"To",
|
||
"Subject",
|
||
"Message-ID",
|
||
"Date",
|
||
"X-Priority",
|
||
"Importance",
|
||
"X-Auto-Response-Suppress",
|
||
"List-Unsubscribe",
|
||
"List-Unsubscribe-Post",
|
||
):
|
||
value = message.get(header)
|
||
if value is not None:
|
||
encrypted_message[header] = value
|
||
|
||
# Required PGP/MIME version part.
|
||
version_part = MIMEApplication(
|
||
b"Version: 1\r\n",
|
||
_subtype="pgp-encrypted",
|
||
)
|
||
version_part.add_header(
|
||
"Content-Description",
|
||
"PGP/MIME version identification",
|
||
)
|
||
encrypted_message.attach(version_part)
|
||
|
||
# Encrypted MIME payload.
|
||
encrypted_part = MIMEApplication(
|
||
encrypted_bytes,
|
||
_subtype="octet-stream",
|
||
)
|
||
encrypted_part.add_header(
|
||
"Content-Description",
|
||
"OpenPGP encrypted message",
|
||
)
|
||
encrypted_part.add_header(
|
||
"Content-Disposition",
|
||
"inline",
|
||
filename="encrypted.asc",
|
||
)
|
||
encrypted_message.attach(encrypted_part)
|
||
|
||
return encrypted_message
|
||
|
||
|
||
async def create_email_message(
|
||
to_email: str,
|
||
subject: str,
|
||
html_content: str,
|
||
user_id: Optional[int] = None,
|
||
):
|
||
"""
|
||
Create an HTML email.
|
||
|
||
If the recipient has a public GPG key in the database, the complete
|
||
MIME message is encrypted using PGP/MIME.
|
||
|
||
Expected database structure:
|
||
|
||
{
|
||
"email": "recipient@example.com",
|
||
"gnupg": "-----BEGIN PGP PUBLIC KEY BLOCK-----\\n..."
|
||
}
|
||
"""
|
||
|
||
msg = MIMEMultipart("alternative")
|
||
|
||
msg["From"] = f"{FROM_NAME} <{FROM_EMAIL}>"
|
||
msg["To"] = to_email
|
||
msg["Subject"] = subject
|
||
msg["Message-ID"] = make_msgid(
|
||
domain=FROM_EMAIL.rsplit("@", 1)[1],
|
||
)
|
||
msg["Date"] = formatdate(localtime=True)
|
||
msg["X-Priority"] = "3"
|
||
msg["Importance"] = "normal"
|
||
msg["X-Auto-Response-Suppress"] = "All"
|
||
|
||
# Do not send unsubscribe headers to yourself.
|
||
if to_email.lower() != FROM_EMAIL.lower():
|
||
unsubscribe_url = build_unsubscribe_url(
|
||
to_email=to_email,
|
||
user_id=user_id,
|
||
domain=FULL_DOMAIN,
|
||
)
|
||
|
||
msg["List-Unsubscribe"] = f"<{unsubscribe_url}>"
|
||
msg["List-Unsubscribe-Post"] = "List-Unsubscribe=One-Click"
|
||
|
||
msg.attach(
|
||
MIMEText(
|
||
html_content,
|
||
"html",
|
||
"utf-8",
|
||
)
|
||
)
|
||
|
||
# Fetch the recipient's public key.
|
||
user = await users_collection.find_one(
|
||
{"email": to_email},
|
||
{"gnupg": 1},
|
||
)
|
||
|
||
public_key = user.get("gnupg") if user else None
|
||
|
||
if not public_key:
|
||
# Return a normal unencrypted email when no key is available.
|
||
return msg
|
||
|
||
if not isinstance(public_key, str):
|
||
raise TypeError("The gnupg database field must be a string")
|
||
|
||
try:
|
||
fingerprint = import_public_key(public_key)
|
||
return encrypt_mime_message(msg, fingerprint)
|
||
|
||
except Exception:
|
||
# Encryption is mandatory whenever a key exists.
|
||
# Do not silently send the message unencrypted.
|
||
raise
|
||
|
||
|
||
def connect(email_server, email_port):
|
||
server = smtplib.SMTP(email_server, email_port)
|
||
server.starttls(context=ctx)
|
||
return server
|
||
|
||
|
||
def send_email(
|
||
to_email: str,
|
||
html_content: str,
|
||
subject: str,
|
||
user_id: Optional[str] = None,
|
||
) -> bool:
|
||
msg =asyncio.run(create_email_message(
|
||
to_email=to_email,
|
||
subject=subject,
|
||
html_content=html_content,
|
||
user_id=user_id,
|
||
))
|
||
|
||
try:
|
||
with connect(EMAIL_SMTP_SERVER, EMAIL_SMTP_PORT) as server:
|
||
server.login(FROM_EMAIL, EMAIL_PASS)
|
||
server.send_message(msg)
|
||
# server.close()
|
||
|
||
logging.info(f"Email sent successfully to: {to_email} | Subject: {subject}")
|
||
return True
|
||
|
||
except Exception as e:
|
||
logging.error(
|
||
f"Error sending email to: {to_email}, "
|
||
f"Subject: {subject}, Error: {str(e)}"
|
||
)
|
||
return False
|
||
|
||
|
||
celery_app_email = Celery('emails', broker=f'{redis_uri}2', backend=f'{redis_uri}2', task_track_started=True)
|
||
|
||
|
||
@celery_app_email.task(
|
||
bind=True,
|
||
rate_limit='30/m',
|
||
name='email_task.send_email',
|
||
retry_backoff=True,
|
||
retry_backoff_max=600,
|
||
retry_jitter=True,
|
||
max_retries=3,
|
||
soft_time_limit=30,
|
||
time_limit=45,
|
||
)
|
||
def notify(self, to_email: str, data: dict, subject: str = EMAIL_SUBJECT, template: str = 'main.html'):
|
||
logging.info(f"Отправка email на {to_email} (попытка {self.request.retries + 1})")
|
||
try:
|
||
html_content = templates.get_template(f"email/{template}").render({
|
||
"subject": subject,
|
||
"data": data
|
||
})
|
||
|
||
ok = send_email(to_email, html_content, subject=subject)
|
||
if ok:
|
||
return ok
|
||
else:
|
||
raise self.retry(exc=exc)
|
||
|
||
except Exception as exc:
|
||
logging.warning(
|
||
f"Ошибка при отправке email на {to_email} (попытка {self.request.retries + 1}): {exc}"
|
||
)
|
||
raise self.retry(exc=exc)
|