refactor(i-g): BUG-018 Pilot — mail/services.py Split Schritt 1 (crypto+sanitize+pgp extrahiert)

Die 3 pure-function Bloecke aus services.py in eigene Sub-Module extrahiert: crypto.py (AES-256 Fernet mit Legacy-Salt + MAIL_ENCRYPTION_KEY-Guard), sanitize.py (nh3 HTML-Sanitizer), pgp.py (pgpy-basiert). Rueckwaertskompatibilitaet via Re-Export-Imports in services.py — alle 4 Consumer unveraendert. Beweis: mail+sig_label_routes 51/51 passed in 105.52s; ruff clean.
2026-08-26 09:29:47 +02:00
parent 3e43219b84
commit a1d5e56009
4 changed files with 200 additions and 160 deletions
@@ -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()
@@ -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")
@@ -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"},
+        },
+    )
@@ -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) ───