Files
gh-christianlouis-docuelevate/app/utils/encryption.py
T
copilot-swe-agent[bot] d08040ac4a style: fix all flake8 linter errors across app/ directory
- Run Black formatter and isort on all app/ files
- Remove unused imports (F401) across multiple files
- Add # noqa: F401 for intentional re-exports in celery_worker.py,
  tasks/__init__.py, utils.py, frontend.py, views/base.py
- Fix f-strings without placeholders (F541) in azure.py, notification.py,
  check_credentials.py, upload_to_onedrive.py, settings.py
- Fix bare except (E722) in upload_to_sftp.py
- Fix block comment format (E265) in models.py
- Move imports to top of file to fix E402 in celery_app.py, celery_worker.py
- Fix line-too-long (E501) by wrapping strings in multiple files
- Remove unused variable (F841) in upload_to_nextcloud.py

Co-authored-by: christianlouis <361235+christianlouis@users.noreply.github.com>
2026-02-08 17:42:33 +00:00

149 lines
4.2 KiB
Python

"""
Encryption utilities for securing sensitive settings in the database.
Uses Fernet symmetric encryption with a key derived from SESSION_SECRET.
This provides encryption at rest for sensitive configuration values.
"""
import base64
import hashlib
import logging
from typing import Optional
logger = logging.getLogger(__name__)
# Lazy-load cryptography to avoid import errors if not installed
_cipher_suite = None
def _get_cipher_suite():
"""
Get or create the Fernet cipher suite for encryption/decryption.
The encryption key is derived from SESSION_SECRET to ensure:
1. Settings are encrypted at rest in the database
2. The same key is used across app restarts
3. No additional secret management needed
Returns:
Fernet cipher suite instance
"""
global _cipher_suite
if _cipher_suite is None:
try:
from cryptography.fernet import Fernet
from app.config import settings
# Derive a Fernet-compatible key from SESSION_SECRET
# Fernet requires a 32-byte base64-encoded key
secret = settings.session_secret.encode("utf-8")
# Use SHA256 to get exactly 32 bytes, then base64 encode
key_bytes = hashlib.sha256(secret).digest()
fernet_key = base64.urlsafe_b64encode(key_bytes)
_cipher_suite = Fernet(fernet_key)
logger.debug("Encryption cipher suite initialized")
except ImportError:
logger.warning(
"cryptography library not installed. "
"Sensitive settings will be stored in plaintext. "
"Install with: pip install cryptography"
)
_cipher_suite = None
except Exception as e:
logger.error(f"Failed to initialize encryption: {e}")
_cipher_suite = None
return _cipher_suite
def encrypt_value(plaintext: Optional[str]) -> Optional[str]:
"""
Encrypt a plaintext value for storage in the database.
Args:
plaintext: The value to encrypt (or None)
Returns:
Encrypted value as base64 string, or plaintext if encryption unavailable
"""
if plaintext is None or plaintext == "":
return plaintext
cipher = _get_cipher_suite()
if cipher is None:
# Encryption not available, store in plaintext with warning
logger.warning("Storing sensitive value in plaintext (encryption unavailable)")
return plaintext
try:
encrypted_bytes = cipher.encrypt(plaintext.encode("utf-8"))
# Prefix with "enc:" to identify encrypted values
return "enc:" + encrypted_bytes.decode("utf-8")
except Exception as e:
logger.error(f"Encryption failed: {e}")
# Fall back to plaintext
return plaintext
def decrypt_value(ciphertext: Optional[str]) -> Optional[str]:
"""
Decrypt a value from the database.
Args:
ciphertext: The encrypted value (or plaintext if not encrypted)
Returns:
Decrypted plaintext value
"""
if ciphertext is None or ciphertext == "":
return ciphertext
# Check if value is encrypted (has "enc:" prefix)
if not ciphertext.startswith("enc:"):
# Not encrypted, return as-is
return ciphertext
cipher = _get_cipher_suite()
if cipher is None:
logger.error("Cannot decrypt value: encryption not available")
return "[ENCRYPTED - Cannot decrypt]"
try:
# Remove "enc:" prefix and decrypt
encrypted_bytes = ciphertext[4:].encode("utf-8")
plaintext_bytes = cipher.decrypt(encrypted_bytes)
return plaintext_bytes.decode("utf-8")
except Exception as e:
logger.error(f"Decryption failed: {e}")
return "[DECRYPTION FAILED]"
def is_encrypted(value: Optional[str]) -> bool:
"""
Check if a value is encrypted.
Args:
value: The value to check
Returns:
True if the value is encrypted, False otherwise
"""
return value is not None and isinstance(value, str) and value.startswith("enc:")
def is_encryption_available() -> bool:
"""
Check if encryption is available.
Returns:
True if cryptography library is installed and encryption is working
"""
return _get_cipher_suite() is not None