"""System settings router - read/update settings, upload company logo.""" from fastapi import APIRouter, Depends, File, HTTPException, UploadFile, status from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from src.backend.config import settings from src.backend.database import get_db from src.backend.api.middleware.api_key import get_current_user, require_admin_user from src.backend.api.routers.files import ALLOWED_IMAGE_TYPES, detect_file_type from src.backend.models.orm.setting import SystemSetting from src.backend.models.orm.user import User router = APIRouter(prefix="/api/settings", tags=["settings"]) # Domains for the settings keys the backend actually reads. Values outside these # would only fail later (csv.writer TypeError on close_run, etc) — reject up front. _VALID_CSV_DELIMITERS = {",", ";", "\t", "|"} _VALID_CSV_DECIMAL_SEPARATORS = {",", "."} def _validate_setting_value(key: str, value: str) -> None: """Raise 422 for a value that would break the code reading this key. Unrecognised keys pass through unvalidated: the settings store is generic. """ if key == "csv_delimiter": if value not in _VALID_CSV_DELIMITERS: raise HTTPException( status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail="csv_delimiter deve essere un carattere singolo fra ',' ';' TAB '|'", ) elif key == "csv_decimal_separator": if value not in _VALID_CSV_DECIMAL_SEPARATORS: raise HTTPException( status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail="csv_decimal_separator deve essere ',' o '.'", ) elif key == "auto_logout_minutes": if value != "" and (not value.isdigit() or not (1 <= int(value) <= 480)): raise HTTPException( status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail="auto_logout_minutes deve essere vuoto (disabilitato) o un intero fra 1 e 480", ) elif key in ("company_name", "company_logo_path"): if not value.strip() or len(value) > 200: raise HTTPException( status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail=f"{key} deve essere una stringa non vuota di massimo 200 caratteri", ) @router.get("/") async def get_settings( user: User = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): """Get all system settings as a dictionary. Returns key-value pairs of all settings. Available to all authenticated users. """ result = await db.execute(select(SystemSetting)) settings_list = result.scalars().all() settings_dict = { setting.setting_key: setting.setting_value for setting in settings_list } return settings_dict @router.put("/") async def update_settings( settings_data: dict[str, str], user: User = Depends(require_admin_user), db: AsyncSession = Depends(get_db), ): """Update multiple system settings. Expects a JSON object with setting_key: setting_value pairs. Only admins can update settings. """ updated_keys = [] for key, value in settings_data.items(): _validate_setting_value(key, value) # Check if setting exists result = await db.execute( select(SystemSetting).where(SystemSetting.setting_key == key) ) setting = result.scalar_one_or_none() if setting is None: # Create new setting (assume type 'string' for new settings) setting = SystemSetting( setting_key=key, setting_value=value, setting_type="string", updated_by=user.id, ) db.add(setting) else: # Update existing setting setting.setting_value = value setting.updated_by = user.id updated_keys.append(key) await db.flush() return { "message": f"Updated {len(updated_keys)} settings", "updated_keys": updated_keys, } @router.post("/logo") async def upload_company_logo( file: UploadFile = File(...), user: User = Depends(require_admin_user), db: AsyncSession = Depends(get_db), ): """Upload company logo. Saves logo to uploads/logos/ and updates company_logo_path setting. Only admins can upload the logo. The type is determined from the file's magic bytes, not the client-declared Content-Type (#36 point 2). SVG is refused outright: it can itself carry script content, and there is no safe way to serve it inline for a logo shown on every page. """ # Read in chunks, stopping as soon as the size limit is crossed instead # of buffering an oversized body into memory first. max_bytes = settings.max_upload_size_mb * 1024 * 1024 buffer = bytearray() chunk_size = 1024 * 1024 while True: chunk = await file.read(chunk_size) if not chunk: break buffer.extend(chunk) if len(buffer) > max_bytes: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail=f"File size exceeds maximum {settings.max_upload_size_mb}MB", ) content = bytes(buffer) file_size = len(content) # Validate file type from the actual bytes (SVG never matches: refused) detected = detect_file_type(content) if detected is None or detected[0] not in ALLOWED_IMAGE_TYPES: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="File type not allowed. Must be a recognised image (SVG is not accepted).", ) _detected_mime, detected_ext = detected # Create logos directory logos_dir = settings.upload_path / "logos" logos_dir.mkdir(parents=True, exist_ok=True) # Fixed filename for the company logo, extension forced to what was # actually detected (not what the client claimed or the filename said). logo_filename = f"company_logo{detected_ext}" logo_path = logos_dir / logo_filename # Write file logo_path.write_bytes(content) # Update setting in database relative_path = logo_path.relative_to(settings.upload_path) logo_path_str = str(relative_path).replace("\\", "/") result = await db.execute( select(SystemSetting).where(SystemSetting.setting_key == "company_logo_path") ) setting = result.scalar_one_or_none() if setting is None: setting = SystemSetting( setting_key="company_logo_path", setting_value=logo_path_str, setting_type="string", description="Path to company logo file", updated_by=user.id, ) db.add(setting) else: setting.setting_value = logo_path_str setting.updated_by = user.id await db.flush() return { "message": "Company logo uploaded successfully", "logo_path": logo_path_str, "file_size": file_size, }