refactor(i-g): BUG-018 Pilot — mail/services.py Split Schritt 1 (crypto+sanitize+pgp extrahiert)
Check Cross-Plugin Imports / check (push) Has been cancelled
Check Cross-Plugin Imports / check (push) Has been cancelled
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.
This commit is contained in:
@@ -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()
|
||||
Reference in New Issue
Block a user