kultmost/avtopost.kultmost.ru
SHA256
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

gnupg

Browse Source
This commit is contained in:
vladislav 2026-09-14 06:50:05 -04:00
parent 26cc09f8e9
commit fe497b5132
4 changed files with 452 additions and 55 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

274
email_task.ipynb
View File

@ -38,7 +38,8 @@
" load_env('env')\n", " load_env('env')\n",
"except:\n", "except:\n",
" load_env('env_example')" " load_env('env_example')"
] ],
"display_name": ".venv"
}, },
{ {
"cell_type": "code", "cell_type": "code",
@ -48,17 +49,33 @@
"outputs": [], "outputs": [],
"source": [ "source": [
"import smtplib\n", "import smtplib\n",
"import os\n",
"from email.mime.application import MIMEApplication\n",
"from email.mime.multipart import MIMEMultipart\n", "from email.mime.multipart import MIMEMultipart\n",
"from email.mime.text import MIMEText\n", "from email.mime.text import MIMEText\n",
"from email.utils import formatdate, make_msgid\n",
"from typing import Optional\n",
"import gnupg\n",
"from celery import Celery\n", "from celery import Celery\n",
"from config import *\n", "from config import *\n",
"from db import *\n", "from db import *\n",
"from templates import templates\n", "from templates import templates\n",
"import logging\n", "import logging\n",
"from typing import Optional\n", "import asyncio \n"
"import asyncio \n", ],
"from email.utils import formatdate, make_msgid\n" "display_name": ".venv"
] },
{
"cell_type": "code",
"execution_count": null,
"id": "ea1066a5",
"metadata": {},
"outputs": [],
"source": [
"def build_unsubscribe_url(email: str, token: str, base_url: str) -> str:\n",
" return f\"{base_url}/api/v1/email/unsubscribe?token={token}&email={email}\"\n"
],
"display_name": ".venv"
}, },
{ {
"cell_type": "code", "cell_type": "code",
@ -67,37 +84,211 @@
"metadata": {}, "metadata": {},
"outputs": [], "outputs": [],
"source": [ "source": [
"def build_unsubscribe_url(email: str, token: str, base_url: str) -> str:\n", "GPG_HOME = os.getenv(\"GPG_HOME\", \"/var/lib/app/.gnupg\")\n",
" return f\"{base_url}/api/v1/email/unsubscribe?token={token}&email={email}\"\n", "GPG_BINARY = os.getenv(\"GPG_BINARY\", \"gpg\")\n",
"\n",
"gpg = gnupg.GPG(\n",
" gnupghome=GPG_HOME,\n",
" gpgbinary=GPG_BINARY,\n",
")\n",
"\n",
"\n",
"def build_unsubscribe_url(\n",
" to_email: str,\n",
" user_id: Optional[int],\n",
" domain: str,\n",
") -> str:\n",
" \"\"\"\n",
" Replace this implementation with your real unsubscribe URL builder.\n",
" Avoid putting a raw email address in the URL in production.\n",
" \"\"\"\n",
" from urllib.parse import quote\n",
"\n",
" return f\"https://{domain}/unsubscribe?email={quote(to_email)}\"\n",
"\n",
"\n",
"def import_public_key(public_key: str) -> str:\n",
" \"\"\"\n",
" Import an ASCII-armored public key and return its fingerprint.\n",
"\n",
" The database value should contain the complete public key, for example:\n",
"\n",
" -----BEGIN PGP PUBLIC KEY BLOCK-----\n",
" ...\n",
" -----END PGP PUBLIC KEY BLOCK-----\n",
" \"\"\"\n",
" result = gpg.import_keys(public_key)\n",
"\n",
" if not result.fingerprints:\n",
" raise RuntimeError(\n",
" f\"Unable to import the recipient's public key: {result}\"\n",
" )\n",
"\n",
" # Use the imported fingerprint rather than the email address. The email\n",
" # address in the key's UID may not exactly match the recipient address.\n",
" return result.fingerprints[0]\n",
"\n",
"\n",
"def encrypt_mime_message(\n",
" message,\n",
" recipient_fingerprint: str,\n",
"):\n",
" \"\"\"\n",
" Encrypt an already-built MIME message and return a PGP/MIME message.\n",
" \"\"\"\n",
"\n",
" # Serialize the original MIME message before encryption.\n",
" plaintext = message.as_bytes()\n",
"\n",
" # Binary OpenPGP data is appropriate for application/octet-stream.\n",
" encrypted = gpg.encrypt(\n",
" plaintext,\n",
" recipients=[recipient_fingerprint],\n",
" armor=False,\n",
" always_trust=False,\n",
" )\n",
"\n",
" if not encrypted.ok:\n",
" raise RuntimeError(\n",
" f\"GPG encryption failed: {encrypted.status}; \"\n",
" f\"stderr: {encrypted.stderr}\"\n",
" )\n",
"\n",
" encrypted_bytes = encrypted.data\n",
"\n",
" if not encrypted_bytes:\n",
" raise RuntimeError(\"GPG returned an empty encrypted message\")\n",
"\n",
" encrypted_message = MIMEMultipart(\n",
" \"encrypted\",\n",
" protocol=\"application/pgp-encrypted\",\n",
" )\n",
"\n",
" # Copy normal mail headers. Content-Type and MIME-Version are generated\n",
" # automatically for the new multipart/encrypted message.\n",
" for header in (\n",
" \"From\",\n",
" \"To\",\n",
" \"Subject\",\n",
" \"Message-ID\",\n",
" \"Date\",\n",
" \"X-Priority\",\n",
" \"Importance\",\n",
" \"X-Auto-Response-Suppress\",\n",
" \"List-Unsubscribe\",\n",
" \"List-Unsubscribe-Post\",\n",
" ):\n",
" value = message.get(header)\n",
" if value is not None:\n",
" encrypted_message[header] = value\n",
"\n",
" # Required PGP/MIME version part.\n",
" version_part = MIMEApplication(\n",
" b\"Version: 1\\r\\n\",\n",
" _subtype=\"pgp-encrypted\",\n",
" )\n",
" version_part.add_header(\n",
" \"Content-Description\",\n",
" \"PGP/MIME version identification\",\n",
" )\n",
" encrypted_message.attach(version_part)\n",
"\n",
" # Encrypted MIME payload.\n",
" encrypted_part = MIMEApplication(\n",
" encrypted_bytes,\n",
" _subtype=\"octet-stream\",\n",
" )\n",
" encrypted_part.add_header(\n",
" \"Content-Description\",\n",
" \"OpenPGP encrypted message\",\n",
" )\n",
" encrypted_part.add_header(\n",
" \"Content-Disposition\",\n",
" \"inline\",\n",
" filename=\"encrypted.asc\",\n",
" )\n",
" encrypted_message.attach(encrypted_part)\n",
"\n",
" return encrypted_message\n",
"\n", "\n",
"\n", "\n",
"async def create_email_message(\n", "async def create_email_message(\n",
" to_email: str,\n", " to_email: str,\n",
" subject: str,\n", " subject: str,\n",
" html_content: str,\n", " html_content: str,\n",
" user_id: int = None,\n", " user_id: Optional[int] = None,\n",
"):\n", "):\n",
" \"\"\"\n",
" Create an HTML email.\n",
"\n",
" If the recipient has a public GPG key in the database, the complete\n",
" MIME message is encrypted using PGP/MIME.\n",
"\n",
" Expected database structure:\n",
"\n",
" {\n",
" \"email\": \"recipient@example.com\",\n",
" \"gnupg\": \"-----BEGIN PGP PUBLIC KEY BLOCK-----\\\\n...\"\n",
" }\n",
" \"\"\"\n",
"\n",
" msg = MIMEMultipart(\"alternative\")\n", " msg = MIMEMultipart(\"alternative\")\n",
" \n", "\n",
" msg['From'] = f\"{FROM_NAME} <{FROM_EMAIL}>\"\n", " msg[\"From\"] = f\"{FROM_NAME} <{FROM_EMAIL}>\"\n",
" msg['To'] = to_email\n", " msg[\"To\"] = to_email\n",
" msg['Subject'] = subject\n", " msg[\"Subject\"] = subject\n",
" msg['Message-ID'] = make_msgid(domain=FROM_EMAIL.split('@')[1])\n", " msg[\"Message-ID\"] = make_msgid(\n",
" msg['Date'] = formatdate(localtime=True)\n", " domain=FROM_EMAIL.rsplit(\"@\", 1)[1],\n",
" msg['X-Priority'] = \"3\"\n", " )\n",
" msg['Importance'] = \"normal\"\n", " msg[\"Date\"] = formatdate(localtime=True)\n",
" msg['X-Auto-Response-Suppress'] = \"All\"\n", " msg[\"X-Priority\"] = \"3\"\n",
" \n", " msg[\"Importance\"] = \"normal\"\n",
" # token = await create_unsubscribe_token(to_email, user_id)\n", " msg[\"X-Auto-Response-Suppress\"] = \"All\"\n",
" unsubscribe_url = build_unsubscribe_url(to_email, None, FULL_DOMAIN)\n", "\n",
" if to_email != FROM_EMAIL:\n", " # Do not send unsubscribe headers to yourself.\n",
" msg['List-Unsubscribe'] = f\"<{unsubscribe_url}>\"\n", " if to_email.lower() != FROM_EMAIL.lower():\n",
" msg['List-Unsubscribe-Post'] = \"List-Unsubscribe=One-Click\"\n", " unsubscribe_url = build_unsubscribe_url(\n",
" \n", " to_email=to_email,\n",
" msg.attach(MIMEText(html_content, 'html'))\n", " user_id=user_id,\n",
" \n", " domain=FULL_DOMAIN,\n",
" return msg\n" " )\n",
] "\n",
" msg[\"List-Unsubscribe\"] = f\"<{unsubscribe_url}>\"\n",
" msg[\"List-Unsubscribe-Post\"] = \"List-Unsubscribe=One-Click\"\n",
"\n",
" msg.attach(\n",
" MIMEText(\n",
" html_content,\n",
" \"html\",\n",
" \"utf-8\",\n",
" )\n",
" )\n",
"\n",
" # Fetch the recipient's public key.\n",
" user = await users_collection.find_one(\n",
" {\"email\": to_email},\n",
" {\"gnupg\": 1},\n",
" )\n",
"\n",
" public_key = user.get(\"gnupg\") if user else None\n",
"\n",
" if not public_key:\n",
" # Return a normal unencrypted email when no key is available.\n",
" return msg\n",
"\n",
" if not isinstance(public_key, str):\n",
" raise TypeError(\"The gnupg database field must be a string\")\n",
"\n",
" try:\n",
" fingerprint = import_public_key(public_key)\n",
" return encrypt_mime_message(msg, fingerprint)\n",
"\n",
" except Exception:\n",
" # Encryption is mandatory whenever a key exists.\n",
" # Do not silently send the message unencrypted.\n",
" raise\n"
],
"display_name": ".venv"
}, },
{ {
"cell_type": "code", "cell_type": "code",
@ -140,7 +331,8 @@
" f\"Subject: {subject}, Error: {str(e)}\"\n", " f\"Subject: {subject}, Error: {str(e)}\"\n",
" )\n", " )\n",
" return False\n" " return False\n"
] ],
"display_name": ".venv"
}, },
{ {
"cell_type": "code", "cell_type": "code",
@ -150,7 +342,8 @@
"outputs": [], "outputs": [],
"source": [ "source": [
"celery_app_email = Celery('emails', broker=f'{redis_uri}2', backend=f'{redis_uri}2', task_track_started=True)\n" "celery_app_email = Celery('emails', broker=f'{redis_uri}2', backend=f'{redis_uri}2', task_track_started=True)\n"
] ],
"display_name": ".venv"
}, },
{ {
"cell_type": "code", "cell_type": "code",
@ -189,7 +382,8 @@
" f\"Ошибка при отправке email на {to_email} (попытка {self.request.retries + 1}): {exc}\"\n", " f\"Ошибка при отправке email на {to_email} (попытка {self.request.retries + 1}): {exc}\"\n",
" )\n", " )\n",
" raise self.retry(exc=exc)\n" " raise self.retry(exc=exc)\n"
] ],
"display_name": ".venv"
}, },
{ {
"cell_type": "markdown", "cell_type": "markdown",
@ -199,6 +393,18 @@
"## Run" "## Run"
] ]
}, },
{
"cell_type": "code",
"execution_count": null,
"id": "3aeae7e8",
"metadata": {},
"outputs": [],
"source": [
"!mkdir -p /var/lib/app/.gnupg\n",
"!chmod 700 /var/lib/app/.gnupg"
],
"display_name": ".venv"
},
{ {
"cell_type": "code", "cell_type": "code",
"execution_count": null, "execution_count": null,
@ -228,7 +434,8 @@
"\n", "\n",
" with open(f'{filename_without_extension}.py', 'w') as f:\n", " with open(f'{filename_without_extension}.py', 'w') as f:\n",
" f.write('\\n\\n'.join(app))\n" " f.write('\\n\\n'.join(app))\n"
] ],
"display_name": ".venv"
}, },
{ {
"cell_type": "code", "cell_type": "code",
@ -238,7 +445,8 @@
"outputs": [], "outputs": [],
"source": [ "source": [
"!celery -A email_task.celery_app_email worker --loglevel=info" "!celery -A email_task.celery_app_email worker --loglevel=info"
] ],
"display_name": ".venv"
} }
], ],
"metadata": { "metadata": {

224
email_task.py
View File

@ -32,48 +32,228 @@ except:
load_env('env_example') load_env('env_example')
import smtplib import smtplib
import os
from email.mime.application import MIMEApplication
from email.mime.multipart import MIMEMultipart from email.mime.multipart import MIMEMultipart
from email.mime.text import MIMEText from email.mime.text import MIMEText
from email.utils import formatdate, make_msgid
from typing import Optional
import gnupg
from celery import Celery from celery import Celery
from config import * from config import *
from db import * from db import *
from templates import templates from templates import templates
import logging import logging
from typing import Optional
import asyncio import asyncio
from email.utils import formatdate, make_msgid
def build_unsubscribe_url(email: str, token: str, base_url: str) -> str: def build_unsubscribe_url(email: str, token: str, base_url: str) -> str:
return f"{base_url}/api/v1/email/unsubscribe?token={token}&email={email}" 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( async def create_email_message(
to_email: str, to_email: str,
subject: str, subject: str,
html_content: str, html_content: str,
user_id: int = None, 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 = MIMEMultipart("alternative")
msg['From'] = f"{FROM_NAME} <{FROM_EMAIL}>" msg["From"] = f"{FROM_NAME} <{FROM_EMAIL}>"
msg['To'] = to_email msg["To"] = to_email
msg['Subject'] = subject msg["Subject"] = subject
msg['Message-ID'] = make_msgid(domain=FROM_EMAIL.split('@')[1]) msg["Message-ID"] = make_msgid(
msg['Date'] = formatdate(localtime=True) domain=FROM_EMAIL.rsplit("@", 1)[1],
msg['X-Priority'] = "3" )
msg['Importance'] = "normal" msg["Date"] = formatdate(localtime=True)
msg['X-Auto-Response-Suppress'] = "All" msg["X-Priority"] = "3"
msg["Importance"] = "normal"
# token = await create_unsubscribe_token(to_email, user_id) msg["X-Auto-Response-Suppress"] = "All"
unsubscribe_url = build_unsubscribe_url(to_email, None, FULL_DOMAIN)
if to_email != FROM_EMAIL: # Do not send unsubscribe headers to yourself.
msg['List-Unsubscribe'] = f"<{unsubscribe_url}>" if to_email.lower() != FROM_EMAIL.lower():
msg['List-Unsubscribe-Post'] = "List-Unsubscribe=One-Click" unsubscribe_url = build_unsubscribe_url(
to_email=to_email,
msg.attach(MIMEText(html_content, 'html')) user_id=user_id,
domain=FULL_DOMAIN,
return msg )
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): def connect(email_server, email_port):

1
requirements.txt
View File

@ -38,3 +38,4 @@ celery[redis]==5.6.3
pika==1.3.2 pika==1.3.2
pillow==12.1.1 pillow==12.1.1
pytz pytz
python-gnupg

8
templates/profile/form.html
View File

@ -63,6 +63,12 @@
<span></span> <span></span>
</div> </div>
<div class="c_nb6gy floating-group">
<label for="gnupg" class="c_z3ybu floating-label">GnuPG</label>
<textarea type="text" id="gnupg" name="gnupg" class="c_4ykpn" ></textarea>
<span>Самозащита электронной почты</span>
</div>
<div class="c_aihm6 floating-group"> <div class="c_aihm6 floating-group">
<label for="phone" class="c_z3ybu floating-label">Телефон<span style="color:red">*</span></label> <label for="phone" class="c_z3ybu floating-label">Телефон<span style="color:red">*</span></label>
<input name="phone" id="phone" class="c_xhd5k" required /> <input name="phone" id="phone" class="c_xhd5k" required />
@ -222,6 +228,7 @@ document.getElementById('living_place').value = `{% if user.get('living_place')
document.getElementById('social_media').value = `{% if user.get('social_media') is not none %}{{user.get('social_media','') }}{% endif %}` document.getElementById('social_media').value = `{% if user.get('social_media') is not none %}{{user.get('social_media','') }}{% endif %}`
document.getElementById('marketplaces').value = `{% if user.get('marketplaces') is not none %}{{user.get('marketplaces','') }}{% endif %}` document.getElementById('marketplaces').value = `{% if user.get('marketplaces') is not none %}{{user.get('marketplaces','') }}{% endif %}`
document.getElementById('email').value = `{% if user.get('email') is not none %}{{user.get('email','') }}{% endif %}` document.getElementById('email').value = `{% if user.get('email') is not none %}{{user.get('email','') }}{% endif %}`
document.getElementById('gnupg').value = `{% if user.get('gnupg') is not none %}{{user.get('gnupg','') }}{% endif %}`
document.getElementById('phone').value = `{% if user.get('phone') is not none %}{{user.get('phone','') }}{% endif %}` document.getElementById('phone').value = `{% if user.get('phone') is not none %}{{user.get('phone','') }}{% endif %}`
document.getElementById('messengers').value = `{% if user.get('messengers') is not none %}{{user.get('messengers','') }}{% endif %}` document.getElementById('messengers').value = `{% if user.get('messengers') is not none %}{{user.get('messengers','') }}{% endif %}`
document.getElementById('specialization').value = `{% if user.get('specialization') is not none %}{{user.get('specialization','') }}{% endif %}` document.getElementById('specialization').value = `{% if user.get('specialization') is not none %}{{user.get('specialization','') }}{% endif %}`
@ -270,6 +277,7 @@ document.addEventListener("DOMContentLoaded",()=>{let e=document.getElementById(
'social_media': document.getElementById('social_media').value, 'social_media': document.getElementById('social_media').value,
'marketplaces': document.getElementById('marketplaces').value, 'marketplaces': document.getElementById('marketplaces').value,
'email': document.getElementById('email').value, 'email': document.getElementById('email').value,
'gnupg': document.getElementById('gnupg').value,
'phone': document.getElementById('phone').value, 'phone': document.getElementById('phone').value,
'messengers': document.getElementById('messengers').value, 'messengers': document.getElementById('messengers').value,
'specialization': document.getElementById('specialization').value, 'specialization': document.getElementById('specialization').value,