diff --git a/app/plugins/builtins/mail/crypto.py b/app/plugins/builtins/mail/crypto.py new file mode 100644 index 0000000..d57ee24 --- /dev/null +++ b/app/plugins/builtins/mail/crypto.py @@ -0,0 +1,74 @@ +"""AES-256 password encryption for Mail accounts (Fernet-based). + +Extracted from services.py as part of the God-object split (BUG-018 pilot). +Re-exported by ``app.plugins.builtins.mail.services`` for backwards +compatibility. + +⚠️ The legacy salt and MAIL_ENCRYPTION_KEY env guard are load-bearing: +changing them makes existing encrypted passwords unreadable. +""" + +from __future__ import annotations + +import base64 +import os + +from cryptography.fernet import Fernet +from cryptography.hazmat.primitives import hashes +from cryptography.hazmat.primitives.kdf.pbkdf2 import PBKDF2HMAC + +MAIL_ENCRYPTION_KEY = os.environ.get("MAIL_ENCRYPTION_KEY") +if not MAIL_ENCRYPTION_KEY: + raise RuntimeError( + "MAIL_ENCRYPTION_KEY environment variable is required. " + "Set it to a strong random value." + ) + +# Legacy salt for backward compatibility with existing encrypted passwords +_LEGACY_SALT = b"leocrm-mail-salt" + + +def _derive_key(password: str, salt: bytes) -> bytes: + """Derive a 32-byte Fernet key from a password using PBKDF2 with the given salt.""" + kdf = PBKDF2HMAC( + algorithm=hashes.SHA256(), + length=32, + salt=salt, + iterations=480000, + ) + return base64.urlsafe_b64encode(kdf.derive(password.encode())) + + +def generate_salt() -> str: + """Generate a random 32-byte salt and return as base64 string.""" + salt = os.urandom(32) + return base64.urlsafe_b64encode(salt).decode() + + +def _get_fernet(salt_b64: str | None = None) -> Fernet: + """Get a Fernet instance. If salt_b64 is provided, use it; otherwise use legacy salt.""" + if salt_b64: + salt = base64.urlsafe_b64decode(salt_b64.encode()) + else: + salt = _LEGACY_SALT + return Fernet(_derive_key(MAIL_ENCRYPTION_KEY, salt)) + + +def encrypt_password(plaintext: str, salt_b64: str | None = None) -> str: + """Encrypt a password using AES-256 (Fernet). Returns base64 ciphertext. + + If salt_b64 is provided, uses that salt for key derivation. + If not, uses the legacy hardcoded salt (for backward compatibility). + """ + fernet = _get_fernet(salt_b64) + return fernet.encrypt(plaintext.encode()).decode() + + +def decrypt_password(ciphertext: str, salt_b64: str | None = None) -> str: + """Decrypt a password encrypted with encrypt_password. + + If salt_b64 is provided, uses that salt for key derivation. + If not, uses the legacy hardcoded salt (for backward compatibility). + """ + fernet = _get_fernet(salt_b64) + return fernet.decrypt(ciphertext.encode()).decode() diff --git a/app/plugins/builtins/mail/pgp.py b/app/plugins/builtins/mail/pgp.py new file mode 100644 index 0000000..13a7dae --- /dev/null +++ b/app/plugins/builtins/mail/pgp.py @@ -0,0 +1,48 @@ +"""PGP encryption/decryption for the Mail plugin using pgpy. + +Extracted from services.py as part of the God-object split (BUG-018 pilot). +Re-exported by ``app.plugins.builtins.mail.services``. +""" + +from __future__ import annotations + +import pgpy + + +def import_pgp_private_key(private_key_armored: str, passphrase: str = "") -> tuple[str, str]: + """Import a PGP private key. Returns (key_id, public_key_armored).""" + key, _ = pgpy.PGPKey.from_blob(private_key_armored) + if key.is_protected: + with key.unlock(passphrase): + pub_key = key.pubkey + key_id = str(key.fingerprint).upper()[-16:] + return key_id, str(pub_key) + pub_key = key.pubkey + key_id = str(key.fingerprint).upper()[-16:] + return key_id, str(pub_key) + + +def import_pgp_public_key(public_key_armored: str) -> str: + """Import a PGP public key. Returns key_id.""" + key, _ = pgpy.PGPKey.from_blob(public_key_armored) + return str(key.fingerprint).upper()[-16:] + + +def pgp_encrypt_message(plaintext: str, recipient_public_key_armored: str) -> str: + """Encrypt a message with recipient's public PGP key.""" + pub_key, _ = pgpy.PGPKey.from_blob(recipient_public_key_armored) + msg = pgpy.PGPMessage.new(plaintext) + encrypted = pub_key.encrypt(msg) + return str(encrypted) + + +def pgp_decrypt_message(ciphertext: str, private_key_armored: str, passphrase: str = "") -> str: + """Decrypt a PGP-encrypted message.""" + key, _ = pgpy.PGPKey.from_blob(private_key_armored) + enc_msg = pgpy.PGPMessage.from_blob(ciphertext) + if key.is_protected: + with key.unlock(passphrase): + decrypted = key.decrypt(enc_msg) + return decrypted.message.decode("utf-8") + decrypted = key.decrypt(enc_msg) + return decrypted.message.decode("utf-8") diff --git a/app/plugins/builtins/mail/sanitize.py b/app/plugins/builtins/mail/sanitize.py new file mode 100644 index 0000000..ef9d796 --- /dev/null +++ b/app/plugins/builtins/mail/sanitize.py @@ -0,0 +1,62 @@ +"""HTML sanitization for the Mail plugin using nh3. + +Extracted from services.py as part of the God-object split (BUG-018 pilot). +Re-exported by ``app.plugins.builtins.mail.services``. +""" + +from __future__ import annotations + +import nh3 + + +def sanitize_html(raw_html: str) -> str: + """Sanitize HTML using nh3 — removes script tags and dangerous attributes.""" + if not raw_html: + return "" + return nh3.clean( + raw_html, + tags={ + "a", + "b", + "br", + "div", + "em", + "h1", + "h2", + "h3", + "h4", + "h5", + "h6", + "hr", + "i", + "img", + "li", + "ol", + "p", + "span", + "strong", + "table", + "tbody", + "td", + "th", + "thead", + "tr", + "u", + "ul", + "blockquote", + "code", + "pre", + "font", + "center", + }, + attributes={ + "a": {"href", "title", "target"}, + "img": {"src", "alt", "width", "height"}, + "span": {"style"}, + "div": {"style"}, + "font": {"color", "size", "face"}, + "p": {"style"}, + "td": {"style"}, + "th": {"style"}, + }, + ) diff --git a/app/plugins/builtins/mail/services.py b/app/plugins/builtins/mail/services.py index 0a8d9b3..d13f678 100644 --- a/app/plugins/builtins/mail/services.py +++ b/app/plugins/builtins/mail/services.py @@ -2,7 +2,6 @@ from __future__ import annotations -import base64 import json import logging import os @@ -16,16 +15,26 @@ from email.utils import formataddr, formatdate, make_msgid import aiofiles import aioimaplib import aiosmtplib -import nh3 -import pgpy -from cryptography.fernet import Fernet -from cryptography.hazmat.primitives import hashes -from cryptography.hazmat.primitives.kdf.pbkdf2 import PBKDF2HMAC from sqlalchemy import and_, func, or_, select from sqlalchemy.ext.asyncio import AsyncSession from app.config import settings from app.core.notifications import create_notification + +# ── Re-exports from extracted sub-modules (BUG-018 pilot split) ── +# Backwards compatibility: all consumers still import from +# ``app.plugins.builtins.mail.services`` unchanged. +from app.plugins.builtins.mail.crypto import ( # noqa: E402 + decrypt_password, + encrypt_password, + generate_salt, +) +from app.plugins.builtins.mail.pgp import ( # noqa: E402,F401 + import_pgp_private_key, + import_pgp_public_key, + pgp_decrypt_message, + pgp_encrypt_message, +) from app.plugins.builtins.mail.models import ( Mail, MailAccount, @@ -38,6 +47,7 @@ from app.plugins.builtins.mail.models import ( MailTemplate, VacationSentLog, ) +from app.plugins.builtins.mail.sanitize import sanitize_html # noqa: E402,F401 logger = logging.getLogger(__name__) @@ -129,118 +139,6 @@ def attachment_to_response(att: MailAttachment) -> dict: } -# ─── AES-256 Encryption (Fernet) ─── - -MAIL_ENCRYPTION_KEY = os.environ.get("MAIL_ENCRYPTION_KEY") -if not MAIL_ENCRYPTION_KEY: - raise RuntimeError("MAIL_ENCRYPTION_KEY environment variable is required. Set it to a strong random value.") - -# Legacy salt for backward compatibility with existing encrypted passwords -_LEGACY_SALT = b"leocrm-mail-salt" - - -def _derive_key(password: str, salt: bytes) -> bytes: - """Derive a 32-byte Fernet key from a password using PBKDF2 with the given salt.""" - kdf = PBKDF2HMAC( - algorithm=hashes.SHA256(), - length=32, - salt=salt, - iterations=480000, - ) - return base64.urlsafe_b64encode(kdf.derive(password.encode())) - - -def generate_salt() -> str: - """Generate a random 32-byte salt and return as base64 string.""" - salt = os.urandom(32) - return base64.urlsafe_b64encode(salt).decode() - - -def _get_fernet(salt_b64: str | None = None) -> Fernet: - """Get a Fernet instance. If salt_b64 is provided, use it; otherwise use legacy salt.""" - if salt_b64: - salt = base64.urlsafe_b64decode(salt_b64.encode()) - else: - salt = _LEGACY_SALT - return Fernet(_derive_key(MAIL_ENCRYPTION_KEY, salt)) - - -def encrypt_password(plaintext: str, salt_b64: str | None = None) -> str: - """Encrypt a password using AES-256 (Fernet). Returns base64 ciphertext. - - If salt_b64 is provided, uses that salt for key derivation. - If not, uses the legacy hardcoded salt (for backward compatibility). - """ - fernet = _get_fernet(salt_b64) - return fernet.encrypt(plaintext.encode()).decode() - - -def decrypt_password(ciphertext: str, salt_b64: str | None = None) -> str: - """Decrypt a password encrypted with encrypt_password. - - If salt_b64 is provided, uses that salt for key derivation. - If not, uses the legacy hardcoded salt (for backward compatibility). - """ - fernet = _get_fernet(salt_b64) - return fernet.decrypt(ciphertext.encode()).decode() - - -# ─── HTML Sanitization (F-MAIL: no script tags) ─── - - -def sanitize_html(raw_html: str) -> str: - """Sanitize HTML using nh3 — removes script tags and dangerous attributes.""" - if not raw_html: - return "" - return nh3.clean( - raw_html, - tags={ - "a", - "b", - "br", - "div", - "em", - "h1", - "h2", - "h3", - "h4", - "h5", - "h6", - "hr", - "i", - "img", - "li", - "ol", - "p", - "span", - "strong", - "table", - "tbody", - "td", - "th", - "thead", - "tr", - "u", - "ul", - "blockquote", - "code", - "pre", - "font", - "center", - }, - attributes={ - "a": {"href", "title", "target"}, - "img": {"src", "alt", "width", "height"}, - "span": {"style"}, - "div": {"style"}, - "font": {"color", "size", "face"}, - "p": {"style"}, - "td": {"style"}, - "th": {"style"}, - }, - ) - - # ─── IMAP Quota Parser ─── @@ -1966,48 +1864,6 @@ async def log_vacation_sent( await db.flush() -# ─── PGP Service (F-MAIL-12) ─── - - -def import_pgp_private_key(private_key_armored: str, passphrase: str = "") -> tuple[str, str]: - """Import a PGP private key. Returns (key_id, public_key_armored).""" - key, _ = pgpy.PGPKey.from_blob(private_key_armored) - if key.is_protected: - with key.unlock(passphrase): - pub_key = key.pubkey - key_id = str(key.fingerprint).upper()[-16:] - return key_id, str(pub_key) - pub_key = key.pubkey - key_id = str(key.fingerprint).upper()[-16:] - return key_id, str(pub_key) - - -def import_pgp_public_key(public_key_armored: str) -> str: - """Import a PGP public key. Returns key_id.""" - key, _ = pgpy.PGPKey.from_blob(public_key_armored) - return str(key.fingerprint).upper()[-16:] - - -def pgp_encrypt_message(plaintext: str, recipient_public_key_armored: str) -> str: - """Encrypt a message with recipient's public PGP key.""" - pub_key, _ = pgpy.PGPKey.from_blob(recipient_public_key_armored) - msg = pgpy.PGPMessage.new(plaintext) - encrypted = pub_key.encrypt(msg) - return str(encrypted) - - -def pgp_decrypt_message(ciphertext: str, private_key_armored: str, passphrase: str = "") -> str: - """Decrypt a PGP-encrypted message.""" - key, _ = pgpy.PGPKey.from_blob(private_key_armored) - enc_msg = pgpy.PGPMessage.from_blob(ciphertext) - if key.is_protected: - with key.unlock(passphrase): - decrypted = key.decrypt(enc_msg) - return decrypted.message.decode("utf-8") - decrypted = key.decrypt(enc_msg) - return decrypted.message.decode("utf-8") - - # ─── Contact Linking (F-MAIL-10) ───