126 lines
4.2 KiB
Python
126 lines
4.2 KiB
Python
"""
|
|
utils/crypto.py — Fernet symmetric encryption for website credentials.
|
|
|
|
Key derivation:
|
|
- A 32-byte random salt is generated on first use and stored in config.ini
|
|
under [crypto] / salt.
|
|
- The Fernet key is derived from the salt + a fixed application secret
|
|
using PBKDF2-HMAC-SHA256 (100,000 iterations).
|
|
- This means credentials are tied to the specific config.ini file on the
|
|
operator's machine; moving config.ini to another machine retains access.
|
|
|
|
Migration:
|
|
- _decrypt() tries Fernet first; if that fails it returns the raw value
|
|
unchanged so that plaintext legacy credentials are still readable.
|
|
- Callers should re-encrypt on next write (update_website handles this).
|
|
"""
|
|
|
|
import base64
|
|
import logging
|
|
import os
|
|
import configparser
|
|
|
|
from cryptography.fernet import Fernet, InvalidToken
|
|
from cryptography.hazmat.primitives.kdf.pbkdf2 import PBKDF2HMAC
|
|
from cryptography.hazmat.primitives import hashes
|
|
|
|
logger = logging.getLogger("crypto")
|
|
|
|
_APP_SECRET = b"WebsiteChecker-v1-CredentialKey"
|
|
_CONFIG_FILE = "config.ini"
|
|
_ITERATIONS = 100_000
|
|
_fernet: "Fernet | None" = None
|
|
|
|
|
|
# ─── Key bootstrap ────────────────────────────────────────────────────────────
|
|
|
|
def _get_or_create_salt() -> bytes:
|
|
"""Read salt from config.ini [crypto] section; create and persist if absent."""
|
|
cfg = configparser.ConfigParser()
|
|
cfg.read(_CONFIG_FILE, encoding="utf-8")
|
|
|
|
if "crypto" in cfg and cfg["crypto"].get("salt"):
|
|
return base64.b64decode(cfg["crypto"]["salt"])
|
|
|
|
# Generate a fresh 32-byte salt
|
|
salt = os.urandom(32)
|
|
if "crypto" not in cfg:
|
|
cfg["crypto"] = {}
|
|
cfg["crypto"]["salt"] = base64.b64encode(salt).decode("ascii")
|
|
|
|
with open(_CONFIG_FILE, "w", encoding="utf-8") as fh:
|
|
cfg.write(fh)
|
|
logger.info("Crypto: generated and persisted new credential encryption salt.")
|
|
return salt
|
|
|
|
|
|
def _build_fernet() -> Fernet:
|
|
salt = _get_or_create_salt()
|
|
kdf = PBKDF2HMAC(
|
|
algorithm=hashes.SHA256(),
|
|
length=32,
|
|
salt=salt,
|
|
iterations=_ITERATIONS,
|
|
)
|
|
key = base64.urlsafe_b64encode(kdf.derive(_APP_SECRET))
|
|
return Fernet(key)
|
|
|
|
|
|
def _get_fernet() -> Fernet:
|
|
global _fernet
|
|
if _fernet is None:
|
|
_fernet = _build_fernet()
|
|
return _fernet
|
|
|
|
|
|
def reset_fernet():
|
|
"""Force key reload — call after config.ini is replaced (e.g. settings save)."""
|
|
global _fernet
|
|
_fernet = None
|
|
|
|
|
|
# ─── Public API ───────────────────────────────────────────────────────────────
|
|
|
|
def encrypt(plaintext: str) -> str:
|
|
"""
|
|
Encrypt a plaintext string. Returns a UTF-8-safe ciphertext string
|
|
prefixed with 'enc:' so we can detect encrypted values reliably.
|
|
Returns the original string unchanged if plaintext is empty.
|
|
"""
|
|
if not plaintext:
|
|
return plaintext
|
|
try:
|
|
token = _get_fernet().encrypt(plaintext.encode("utf-8"))
|
|
return "enc:" + token.decode("ascii")
|
|
except Exception as e:
|
|
logger.error(f"Credential encryption failed: {e}")
|
|
return plaintext # safe fallback — don't lose data
|
|
|
|
|
|
def decrypt(ciphertext: str) -> str:
|
|
"""
|
|
Decrypt a ciphertext string produced by encrypt().
|
|
- If ciphertext starts with 'enc:', decrypts with Fernet.
|
|
- Otherwise returns the value as-is (plaintext legacy credential).
|
|
Returns empty string on failure.
|
|
"""
|
|
if not ciphertext:
|
|
return ciphertext
|
|
if not ciphertext.startswith("enc:"):
|
|
# Legacy plaintext — return unchanged; will be re-encrypted on next save
|
|
return ciphertext
|
|
try:
|
|
token = ciphertext[4:].encode("ascii")
|
|
return _get_fernet().decrypt(token).decode("utf-8")
|
|
except InvalidToken:
|
|
logger.error("Credential decryption failed — wrong key or corrupted data.")
|
|
return ""
|
|
except Exception as e:
|
|
logger.error(f"Credential decryption error: {e}")
|
|
return ""
|
|
|
|
|
|
def is_encrypted(value: str) -> bool:
|
|
"""Return True if the value was produced by encrypt()."""
|
|
return isinstance(value, str) and value.startswith("enc:")
|