Add encryption and setup wizard features

ENCRYPTION:
- Add cryptography library for secure storage
- Implement Fernet encryption for sensitive settings
- Key derived from SESSION_SECRET
- Auto-encrypt/decrypt transparent to app
- "enc:" prefix identifies encrypted values
- Graceful fallback if crypto unavailable

SETUP WIZARD:
- Detect fresh installs needing configuration
- 3-step wizard: Infrastructure, Security, AI Services
- "/" redirects to wizard if setup required
- Auto-generate session_secret option
- Skip option for advanced users
- Beautiful UI with progress indicators

UI IMPROVEMENTS:
- Enhanced sensitive field display
- Lock icon showing encryption status
- Improved show/hide toggle for passwords
- Better visual hierarchy

FILES:
- app/utils/encryption.py - Encryption utilities
- app/utils/setup_wizard.py - Wizard detection logic
- app/views/wizard.py - Wizard routes
- frontend/templates/setup_wizard.html - Wizard UI
- requirements.txt - Added cryptography
- IMPLEMENTATION_CHECKLIST.md - Status tracking

Co-authored-by: christianlouis <361235+christianlouis@users.noreply.github.com>
This commit is contained in:
copilot-swe-agent[bot]
2026-02-08 06:18:49 +00:00
parent 121b0d811f
commit 1a01811882
11 changed files with 1020 additions and 206 deletions
+147
View File
@@ -0,0 +1,147 @@
"""
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 logging
import base64
import hashlib
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
+45 -6
View File
@@ -878,16 +878,27 @@ def get_setting_from_db(db: Session, key: str) -> Optional[str]:
"""
Retrieve a setting value from the database.
Automatically decrypts sensitive values if encryption is enabled.
Args:
db: Database session
key: Setting key to retrieve
Returns:
Setting value as string, or None if not found
Setting value as string (decrypted if necessary), or None if not found
"""
try:
setting = db.query(ApplicationSettings).filter(ApplicationSettings.key == key).first()
return setting.value if setting else None
if not setting:
return None
# Check if this setting is sensitive and should be decrypted
metadata = get_setting_metadata(key)
if metadata.get("sensitive", False):
from app.utils.encryption import decrypt_value
return decrypt_value(setting.value)
return setting.value
except SQLAlchemyError as e:
logger.error(f"Error retrieving setting {key} from database: {e}")
return None
@@ -897,6 +908,8 @@ def save_setting_to_db(db: Session, key: str, value: Optional[str]) -> bool:
"""
Save or update a setting in the database.
Automatically encrypts sensitive values if encryption is enabled.
Args:
db: Database session
key: Setting key
@@ -906,11 +919,24 @@ def save_setting_to_db(db: Session, key: str, value: Optional[str]) -> bool:
True if successful, False otherwise
"""
try:
# Check if this setting is sensitive and should be encrypted
metadata = get_setting_metadata(key)
storage_value = value
if metadata.get("sensitive", False) and value:
from app.utils.encryption import encrypt_value, is_encryption_available
if is_encryption_available():
storage_value = encrypt_value(value)
logger.debug(f"Encrypted sensitive setting: {key}")
else:
logger.warning(f"Storing sensitive setting {key} in plaintext (encryption unavailable)")
setting = db.query(ApplicationSettings).filter(ApplicationSettings.key == key).first()
if setting:
setting.value = value
setting.value = storage_value
else:
setting = ApplicationSettings(key=key, value=value)
setting = ApplicationSettings(key=key, value=storage_value)
db.add(setting)
db.commit()
logger.info(f"Saved setting {key} to database")
@@ -925,15 +951,28 @@ def get_all_settings_from_db(db: Session) -> Dict[str, str]:
"""
Retrieve all settings from the database.
Automatically decrypts sensitive values if encryption is enabled.
Args:
db: Database session
Returns:
Dictionary of setting key-value pairs
Dictionary of setting key-value pairs (decrypted)
"""
try:
settings = db.query(ApplicationSettings).all()
return {setting.key: setting.value for setting in settings}
result = {}
for setting in settings:
# Check if this setting is sensitive and should be decrypted
metadata = get_setting_metadata(setting.key)
if metadata.get("sensitive", False):
from app.utils.encryption import decrypt_value
result[setting.key] = decrypt_value(setting.value)
else:
result[setting.key] = setting.value
return result
except SQLAlchemyError as e:
logger.error(f"Error retrieving all settings from database: {e}")
return {}
+212
View File
@@ -0,0 +1,212 @@
"""
Setup wizard utilities for first-time system configuration.
Detects if the system needs initial setup and provides required settings list.
"""
import logging
from typing import List, Dict, Any
from app.config import settings
logger = logging.getLogger(__name__)
def get_required_settings() -> List[Dict[str, Any]]:
"""
Get list of settings that are absolutely required for the system to operate.
Returns:
List of required setting definitions with metadata
"""
return [
{
"key": "database_url",
"label": "Database URL",
"description": "Database connection string (e.g., sqlite:///./app/database.db)",
"type": "string",
"sensitive": False,
"default": "sqlite:///./app/database.db",
"wizard_step": 1,
"wizard_category": "Core Infrastructure"
},
{
"key": "redis_url",
"label": "Redis URL",
"description": "Redis connection for task queue (e.g., redis://localhost:6379/0)",
"type": "string",
"sensitive": False,
"default": "redis://localhost:6379/0",
"wizard_step": 1,
"wizard_category": "Core Infrastructure"
},
{
"key": "workdir",
"label": "Working Directory",
"description": "Directory for temporary file storage and processing",
"type": "string",
"sensitive": False,
"default": "/workdir",
"wizard_step": 1,
"wizard_category": "Core Infrastructure"
},
{
"key": "gotenberg_url",
"label": "Gotenberg URL",
"description": "Gotenberg service URL for document conversion",
"type": "string",
"sensitive": False,
"default": "http://gotenberg:3000",
"wizard_step": 1,
"wizard_category": "Core Infrastructure"
},
{
"key": "session_secret",
"label": "Session Secret",
"description": "Secret key for session encryption (min 32 characters, auto-generate recommended)",
"type": "string",
"sensitive": True,
"default": None, # Should be generated
"wizard_step": 2,
"wizard_category": "Security"
},
{
"key": "admin_username",
"label": "Admin Username",
"description": "Username for the admin account",
"type": "string",
"sensitive": False,
"default": "admin",
"wizard_step": 2,
"wizard_category": "Security"
},
{
"key": "admin_password",
"label": "Admin Password",
"description": "Password for the admin account",
"type": "string",
"sensitive": True,
"default": None, # Must be set
"wizard_step": 2,
"wizard_category": "Security"
},
{
"key": "openai_api_key",
"label": "OpenAI API Key",
"description": "API key for OpenAI services (metadata extraction)",
"type": "string",
"sensitive": True,
"default": None,
"wizard_step": 3,
"wizard_category": "AI Services"
},
{
"key": "azure_ai_key",
"label": "Azure AI Key",
"description": "Azure AI key for document intelligence (OCR)",
"type": "string",
"sensitive": True,
"default": None,
"wizard_step": 3,
"wizard_category": "AI Services"
},
{
"key": "azure_region",
"label": "Azure Region",
"description": "Azure region for AI services (e.g., eastus)",
"type": "string",
"sensitive": False,
"default": "eastus",
"wizard_step": 3,
"wizard_category": "AI Services"
},
{
"key": "azure_endpoint",
"label": "Azure Endpoint",
"description": "Azure AI endpoint URL",
"type": "string",
"sensitive": False,
"default": None,
"wizard_step": 3,
"wizard_category": "AI Services"
},
]
def is_setup_required() -> bool:
"""
Check if the system requires initial setup.
Returns True if any critical required settings are missing or have placeholder values.
Returns:
True if setup wizard should be shown, False otherwise
"""
try:
# Critical settings that must be configured
critical_settings = [
("session_secret", ["INSECURE_DEFAULT_FOR_DEVELOPMENT_ONLY_DO_NOT_USE_IN_PRODUCTION_MINIMUM_32_CHARS"]),
("admin_password", [None, "", "your_secure_password", "changeme", "admin"]),
("openai_api_key", [None, "", "<OPENAI_API_KEY>", "test-key"]),
("azure_ai_key", [None, "", "<AZURE_AI_KEY>", "test-key"]),
]
for setting_key, invalid_values in critical_settings:
value = getattr(settings, setting_key, None)
if value in invalid_values:
logger.warning(f"Setup required: {setting_key} has placeholder or missing value")
return True
# All critical settings are configured
return False
except Exception as e:
logger.error(f"Error checking if setup required: {e}")
# If we can't check, assume setup is not required (fail open)
return False
def get_missing_required_settings() -> List[str]:
"""
Get list of required settings that are missing or have placeholder values.
Returns:
List of setting keys that need to be configured
"""
missing = []
for required_setting in get_required_settings():
key = required_setting["key"]
value = getattr(settings, key, None)
# Check if value is missing or is a placeholder
placeholder_values = [
None, "",
f"<{key.upper()}>",
"test-key",
"your_secure_password",
"changeme",
"INSECURE_DEFAULT_FOR_DEVELOPMENT_ONLY_DO_NOT_USE_IN_PRODUCTION_MINIMUM_32_CHARS"
]
if value in placeholder_values:
missing.append(key)
return missing
def get_wizard_steps() -> Dict[int, List[Dict[str, Any]]]:
"""
Get setup wizard steps organized by step number.
Returns:
Dictionary mapping step number to list of settings in that step
"""
steps = {}
for setting in get_required_settings():
step_num = setting.get("wizard_step", 1)
if step_num not in steps:
steps[step_num] = []
steps[step_num].append(setting)
return steps
+2
View File
@@ -11,9 +11,11 @@ from app.views.dropbox import router as dropbox_router
from app.views.google_drive import router as google_drive_router
from app.views.license_routes import router as license_router # Add the license router
from app.views.settings import router as settings_router
from app.views.wizard import router as wizard_router
# Create a main router that includes all the view routers
router = APIRouter()
router.include_router(wizard_router) # Wizard first (for /setup)
router.include_router(general_router)
router.include_router(status_router)
router.include_router(onedrive_router)
+19 -1
View File
@@ -14,7 +14,25 @@ router = APIRouter()
@router.get("/", include_in_schema=False)
async def serve_index(request: Request, db: Session = Depends(get_db)):
"""Serve the index/home page."""
"""
Serve the index/home page.
If the system requires initial setup, redirect to the setup wizard.
"""
# Check if setup wizard is needed
from app.utils.setup_wizard import is_setup_required
from app.utils.settings_service import get_setting_from_db
# Check if setup was explicitly skipped
setup_skipped = get_setting_from_db(db, "_setup_wizard_skipped")
# Check setup completion query param
setup_complete = request.query_params.get("setup") == "complete"
if not setup_skipped and not setup_complete and is_setup_required():
logger.info("System requires initial setup, redirecting to wizard")
return RedirectResponse(url="/setup?step=1", status_code=303)
# Get provider information from config validator
providers = get_provider_status()
+133
View File
@@ -0,0 +1,133 @@
"""
Setup wizard views for initial system configuration.
"""
import os
import logging
import secrets
from fastapi import Request, Depends, Form
from fastapi.responses import RedirectResponse
from sqlalchemy.orm import Session
from app.views.base import APIRouter, templates, get_db
from app.utils.setup_wizard import (
is_setup_required,
get_required_settings,
get_wizard_steps,
get_missing_required_settings
)
from app.utils.settings_service import save_setting_to_db
logger = logging.getLogger(__name__)
router = APIRouter()
@router.get("/setup")
async def setup_wizard(request: Request, step: int = 1):
"""
Setup wizard for first-time configuration.
This wizard guides users through configuring essential settings
needed for the system to operate properly.
"""
# Get wizard steps
wizard_steps = get_wizard_steps()
max_step = max(wizard_steps.keys())
# Validate step number
if step < 1:
step = 1
elif step > max_step:
step = max_step
# Get settings for current step
current_settings = wizard_steps.get(step, [])
# Get step category (all settings in a step should have same category)
step_category = current_settings[0].get("wizard_category", "Configuration") if current_settings else "Configuration"
return templates.TemplateResponse(
"setup_wizard.html",
{
"request": request,
"current_step": step,
"max_step": max_step,
"settings": current_settings,
"step_category": step_category,
"progress_percent": int((step / max_step) * 100)
}
)
@router.post("/setup")
async def setup_wizard_save(
request: Request,
step: int = Form(...),
db: Session = Depends(get_db)
):
"""
Save settings from the current wizard step.
"""
try:
# Get form data
form_data = await request.form()
# Get settings for current step
wizard_steps = get_wizard_steps()
current_settings = wizard_steps.get(step, [])
# Save each setting from the form
saved_count = 0
for setting in current_settings:
key = setting["key"]
value = form_data.get(key)
# Skip empty values unless it's explicitly allowed
if value and value.strip():
# Auto-generate session_secret if needed
if key == "session_secret" and value == "auto-generate":
value = secrets.token_hex(32)
logger.info("Auto-generated session secret")
# Save to database
if save_setting_to_db(db, key, value):
saved_count += 1
logger.info(f"Setup wizard: Saved {key}")
logger.info(f"Setup wizard step {step}: Saved {saved_count} settings")
# Determine next step
max_step = max(wizard_steps.keys())
next_step = step + 1
if next_step > max_step:
# Setup complete, redirect to home
return RedirectResponse(url="/?setup=complete", status_code=303)
else:
# Go to next step
return RedirectResponse(url=f"/setup?step={next_step}", status_code=303)
except Exception as e:
logger.error(f"Error saving wizard settings: {e}")
return RedirectResponse(url=f"/setup?step={step}&error=save_failed", status_code=303)
@router.get("/setup/skip")
async def setup_wizard_skip(request: Request):
"""
Skip the setup wizard (for advanced users).
Creates a marker to indicate setup was skipped.
"""
try:
db = next(get_db())
try:
# Save a marker to indicate setup was skipped
save_setting_to_db(db, "_setup_wizard_skipped", "true")
logger.info("Setup wizard skipped by user")
return RedirectResponse(url="/", status_code=303)
finally:
db.close()
except Exception as e:
logger.error(f"Error skipping setup wizard: {e}")
return RedirectResponse(url="/", status_code=303)