diff --git a/app/api/dropbox.py b/app/api/dropbox.py index da52c758..fe1b13e9 100644 --- a/app/api/dropbox.py +++ b/app/api/dropbox.py @@ -14,7 +14,7 @@ from app.auth import require_login from app.config import settings from app.database import get_db from app.utils.oauth_helper import exchange_oauth_token -from app.utils.settings_service import save_setting_to_db +from app.utils.settings_service import save_setting_to_db, update_env_file from app.utils.settings_sync import notify_settings_updated # Set up logging @@ -23,6 +23,17 @@ logger = logging.getLogger(__name__) router = APIRouter() +def _require_admin(request: Request) -> dict: + """Dependency to ensure the current user is an admin.""" + user = request.session.get("user") + if not user or not user.get("is_admin"): + raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Admin access required") + return user + + +AdminUser = Annotated[dict, Depends(_require_admin)] + + @router.post("/dropbox/exchange-token") @require_login async def exchange_dropbox_token( @@ -211,9 +222,9 @@ async def test_dropbox_token(request: Request): @router.post("/dropbox/save-settings") -@require_login async def save_dropbox_settings( request: Request, + _admin: AdminUser, refresh_token: Annotated[str, Form(...)], app_key: Annotated[Optional[str], Form()] = None, app_secret: Annotated[Optional[str], Form()] = None, @@ -252,44 +263,16 @@ async def save_dropbox_settings( # Best-effort .env file write try: env_path = os.path.join(os.path.dirname(os.path.dirname(os.path.dirname(__file__))), ".env") - if not os.path.exists(env_path): - logger.warning(f".env file not found at {env_path}, skipping file write") - else: - logger.info(f"Updating Dropbox settings in {env_path}") + dropbox_settings = {"DROPBOX_REFRESH_TOKEN": refresh_token} + if app_key: + dropbox_settings["DROPBOX_APP_KEY"] = app_key + if app_secret: + dropbox_settings["DROPBOX_APP_SECRET"] = app_secret + if folder_path: + dropbox_settings["DROPBOX_FOLDER"] = folder_path - with open(env_path, "r") as f: - env_lines = f.readlines() - - dropbox_settings = {"DROPBOX_REFRESH_TOKEN": refresh_token} - if app_key: - dropbox_settings["DROPBOX_APP_KEY"] = app_key - if app_secret: - dropbox_settings["DROPBOX_APP_SECRET"] = app_secret - if folder_path: - dropbox_settings["DROPBOX_FOLDER"] = folder_path - - updated = set() - new_env_lines = [] - for line in env_lines: - stripped_line = line.rstrip() - is_updated = False - for key, value in dropbox_settings.items(): - if stripped_line.startswith(f"{key}=") or stripped_line.startswith(f"# {key}="): - new_env_lines.append(f"{key}={value}") - updated.add(key) - is_updated = True - break - if not is_updated: - new_env_lines.append(stripped_line) - - for key, value in dropbox_settings.items(): - if key not in updated: - new_env_lines.append(f"{key}={value}") - - with open(env_path, "w") as f: - f.write("\n".join(new_env_lines) + "\n") - - logger.info("Successfully updated Dropbox settings in .env file") + if not update_env_file(env_path, dropbox_settings): + logger.info("Continuing with in-memory update despite .env file update failure or skip") except Exception as env_err: logger.warning(f"Failed to write .env file (non-fatal): {env_err}") diff --git a/app/api/google_drive.py b/app/api/google_drive.py index f9ca4f92..69d52f19 100644 --- a/app/api/google_drive.py +++ b/app/api/google_drive.py @@ -14,7 +14,7 @@ from app.auth import require_login from app.config import settings from app.database import get_db from app.utils.oauth_helper import exchange_oauth_token -from app.utils.settings_service import save_setting_to_db +from app.utils.settings_service import save_setting_to_db, update_env_file from app.utils.settings_sync import notify_settings_updated # Set up logging @@ -23,6 +23,17 @@ logger = logging.getLogger(__name__) router = APIRouter() +def _require_admin(request: Request) -> dict: + """Dependency to ensure the current user is an admin.""" + user = request.session.get("user") + if not user or not user.get("is_admin"): + raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Admin access required") + return user + + +AdminUser = Annotated[dict, Depends(_require_admin)] + + @router.post("/google-drive/exchange-token") @require_login async def exchange_google_drive_token( @@ -362,9 +373,9 @@ def format_time_remaining(time_delta): @router.post("/google-drive/save-settings") -@require_login async def save_google_drive_settings( request: Request, + _admin: AdminUser, refresh_token: Annotated[str, Form(...)], client_id: Annotated[Optional[str], Form()] = None, client_secret: Annotated[Optional[str], Form()] = None, @@ -404,46 +415,9 @@ async def save_google_drive_settings( drive_settings["GOOGLE_DRIVE_FOLDER_ID"] = folder_id # Try to update the .env file, but don't fail if it doesn't exist (for Docker containers) - if os.path.exists(env_path): - try: - logger.info(f"Updating Google Drive settings in {env_path}") - - # Read the current .env file - with open(env_path, "r") as f: - env_lines = f.readlines() - - # Process each line and update or add settings - updated = set() - new_env_lines = [] - for line in env_lines: - stripped_line = line.rstrip() - is_updated = False - for key, value in drive_settings.items(): - if stripped_line.startswith(f"{key}=") or stripped_line.startswith(f"# {key}="): - # Uncomment if commented out - check the original stripped line - new_env_lines.append(f"{key}={value}") - updated.add(key) - is_updated = True - break - if not is_updated: - new_env_lines.append(stripped_line) - - # Add any settings that weren't updated (they weren't in the file) - for key, value in drive_settings.items(): - if key not in updated: - new_env_lines.append(f"{key}={value}") - - # Write the updated .env file - with open(env_path, "w") as f: - f.write("\n".join(new_env_lines) + "\n") - - logger.info("Successfully updated Google Drive settings in .env file") - except Exception as e: - logger.warning(f"Failed to update .env file: {str(e)}, but will continue with in-memory update") - else: - logger.warning( - f".env file not found at {env_path}, skipping file update but continuing with in-memory update" - ) + env_write_success = update_env_file(env_path, drive_settings) + if not env_write_success: + logger.info("Continuing with in-memory update despite .env file update failure or skip") # Update the settings in memory (this always happens) if refresh_token: @@ -481,7 +455,7 @@ async def save_google_drive_settings( return { "status": "success", "message": "Google Drive settings have been saved", - "in_memory_only": not os.path.exists(env_path), + "in_memory_only": not env_write_success, } except Exception as e: diff --git a/app/api/onedrive.py b/app/api/onedrive.py index 25429344..a89103d3 100644 --- a/app/api/onedrive.py +++ b/app/api/onedrive.py @@ -3,6 +3,7 @@ OneDrive API endpoints """ import logging +import os from datetime import datetime, timedelta from typing import Annotated, Optional @@ -13,9 +14,8 @@ from sqlalchemy.orm import Session from app.auth import require_login from app.config import settings from app.database import get_db -from app.utils.env_utils import update_env_file from app.utils.oauth_helper import exchange_oauth_token -from app.utils.settings_service import save_setting_to_db +from app.utils.settings_service import save_setting_to_db, update_env_file from app.utils.settings_sync import notify_settings_updated # Set up logging @@ -24,6 +24,17 @@ logger = logging.getLogger(__name__) router = APIRouter() +def _require_admin(request: Request) -> dict: + """Dependency to ensure the current user is an admin.""" + user = request.session.get("user") + if not user or not user.get("is_admin"): + raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Admin access required") + return user + + +AdminUser = Annotated[dict, Depends(_require_admin)] + + @router.post("/onedrive/exchange-token") @require_login async def exchange_onedrive_token( @@ -204,9 +215,9 @@ def format_time_remaining(time_delta): @router.post("/onedrive/save-settings") -@require_login async def save_onedrive_settings( request: Request, + _admin: AdminUser, refresh_token: Annotated[str, Form(...)], client_id: Annotated[Optional[str], Form()] = None, client_secret: Annotated[Optional[str], Form()] = None, @@ -223,26 +234,44 @@ async def save_onedrive_settings( user.get("preferred_username") or user.get("username") or user.get("email") or user.get("id") or "wizard" ) - # Build settings dictionary mapped to database/memory keys - onedrive_settings = { - "onedrive_refresh_token": refresh_token, - "onedrive_client_id": client_id, - "onedrive_client_secret": client_secret, - "onedrive_tenant_id": tenant_id, - "onedrive_folder_path": folder_path, - } + # Best-effort .env file write + env_path = os.path.join(os.path.dirname(os.path.dirname(os.path.dirname(__file__))), ".env") + onedrive_settings = {"ONEDRIVE_REFRESH_TOKEN": refresh_token} + if client_id: + onedrive_settings["ONEDRIVE_CLIENT_ID"] = client_id + if client_secret: + onedrive_settings["ONEDRIVE_CLIENT_SECRET"] = client_secret + if tenant_id: + onedrive_settings["ONEDRIVE_TENANT_ID"] = tenant_id + if folder_path: + onedrive_settings["ONEDRIVE_FOLDER_PATH"] = folder_path - # Filter out None values - onedrive_settings = {k: v for k, v in onedrive_settings.items() if v is not None} + if not update_env_file(env_path, onedrive_settings): + logger.info("Continuing with in-memory update despite .env file update failure or skip") - # Best-effort .env file write using the new utility - env_settings = {k.upper(): v for k, v in onedrive_settings.items()} - update_env_file(env_settings) + # Update the settings in memory + if refresh_token: + settings.onedrive_refresh_token = refresh_token + if client_id: + settings.onedrive_client_id = client_id + if client_secret: + settings.onedrive_client_secret = client_secret + if tenant_id: + settings.onedrive_tenant_id = tenant_id + if folder_path: + settings.onedrive_folder_path = folder_path - # Update in-memory settings and persist to database dynamically - for key, value in onedrive_settings.items(): - setattr(settings, key, value) - save_setting_to_db(db, key, value, changed_by=changed_by) + # Persist to database (primary) + if refresh_token: + save_setting_to_db(db, "onedrive_refresh_token", refresh_token, changed_by=changed_by) + if client_id: + save_setting_to_db(db, "onedrive_client_id", client_id, changed_by=changed_by) + if client_secret: + save_setting_to_db(db, "onedrive_client_secret", client_secret, changed_by=changed_by) + if tenant_id: + save_setting_to_db(db, "onedrive_tenant_id", tenant_id, changed_by=changed_by) + if folder_path: + save_setting_to_db(db, "onedrive_folder_path", folder_path, changed_by=changed_by) notify_settings_updated() diff --git a/app/utils/env_utils.py b/app/utils/env_utils.py index dfa8ad89..d38a63bf 100644 --- a/app/utils/env_utils.py +++ b/app/utils/env_utils.py @@ -1,11 +1,10 @@ import logging import os -from typing import Dict logger = logging.getLogger(__name__) -def update_env_file(settings_to_update: Dict[str, str]) -> bool: +def update_env_file(settings_to_update: dict[str, str]) -> bool: """ Updates the .env file with the given settings (best-effort). Creates or modifies existing keys. diff --git a/app/utils/settings_service.py b/app/utils/settings_service.py index fef1429d..cec83c24 100644 --- a/app/utils/settings_service.py +++ b/app/utils/settings_service.py @@ -8,6 +8,7 @@ This module provides functionality to: """ import logging +import os from typing import Any, Dict, List, Optional, Tuple from sqlalchemy.exc import SQLAlchemyError @@ -3371,3 +3372,60 @@ def get_settings_for_export(db: Session, source: str = "db") -> Dict[str, str]: # DB only db_settings = get_all_settings_from_db(db) return {k.upper(): v for k, v in sorted(db_settings.items()) if v is not None} + + +def update_env_file(env_path: str, settings_to_update: dict[str, str]) -> bool: + """ + Update an .env file with new settings. + + Reads the file, updates matching settings (even if commented), + appends any that weren't found, and writes the result back. + + Args: + env_path: Path to the .env file + settings_to_update: Dictionary mapping setting names (e.g. 'GOOGLE_DRIVE_USE_OAUTH') to string values + + Returns: + True if the file was successfully updated, False otherwise (e.g. file not found or write error) + """ + if not os.path.exists(env_path): + logger.warning(f".env file not found at {env_path}, skipping file update") + return False + + try: + logger.info(f"Updating settings in {env_path}") + + # Read the current .env file + with open(env_path, "r") as f: + env_lines = f.readlines() + + # Process each line and update or add settings + updated = set() + new_env_lines = [] + for line in env_lines: + stripped_line = line.rstrip() + is_updated = False + for key, value in settings_to_update.items(): + if stripped_line.startswith(f"{key}=") or stripped_line.startswith(f"# {key}="): + # Uncomment if commented out - check the original stripped line + new_env_lines.append(f"{key}={value}") + updated.add(key) + is_updated = True + break + if not is_updated: + new_env_lines.append(stripped_line) + + # Add any settings that weren't updated (they weren't in the file) + for key, value in settings_to_update.items(): + if key not in updated: + new_env_lines.append(f"{key}={value}") + + # Write the updated .env file + with open(env_path, "w") as f: + f.write("\n".join(new_env_lines) + "\n") + + logger.info(f"Successfully updated settings in {env_path}") + return True + except Exception as e: + logger.warning(f"Failed to update {env_path}: {str(e)}") + return False diff --git a/tests/test_api_google_drive_comprehensive.py b/tests/test_api_google_drive_comprehensive.py index 360416a1..6275602f 100644 --- a/tests/test_api_google_drive_comprehensive.py +++ b/tests/test_api_google_drive_comprehensive.py @@ -360,6 +360,15 @@ class TestFormatTimeRemaining: class TestSaveGoogleDriveSettings: """Tests for POST /google-drive/save-settings endpoint.""" + @pytest.fixture(autouse=True) + def _admin_override(self): + from app.api.google_drive import _require_admin + from app.main import app as fastapi_app + + fastapi_app.dependency_overrides[_require_admin] = lambda: {"is_admin": True} + yield + fastapi_app.dependency_overrides.pop(_require_admin, None) + @patch("builtins.open", new_callable=mock_open, read_data="# Existing config\n") @patch("os.path.exists") @patch("os.path.dirname") diff --git a/tests/test_api_google_drive_coverage.py b/tests/test_api_google_drive_coverage.py index 75e092dd..cf242efc 100644 --- a/tests/test_api_google_drive_coverage.py +++ b/tests/test_api_google_drive_coverage.py @@ -195,6 +195,15 @@ class TestGetGoogleDriveTokenInfo: class TestSaveGoogleDriveSettings: """Test save_google_drive_settings endpoint edge cases.""" + @pytest.fixture(autouse=True) + def _admin_override(self): + from app.api.google_drive import _require_admin + from app.main import app as fastapi_app + + fastapi_app.dependency_overrides[_require_admin] = lambda: {"is_admin": True} + yield + fastapi_app.dependency_overrides.pop(_require_admin, None) + @patch("app.api.google_drive.settings") @patch("os.path.exists") def test_save_settings_env_file_not_exists(self, mock_exists, mock_settings, client: TestClient): diff --git a/tests/test_api_google_drive_final.py b/tests/test_api_google_drive_final.py index 49fabdb4..9ef96047 100644 --- a/tests/test_api_google_drive_final.py +++ b/tests/test_api_google_drive_final.py @@ -152,11 +152,16 @@ class TestGetTokenInfoCredentialsBranches: @pytest.mark.unit class TestSaveGoogleDriveSettingsFalsyFields: - """Cover branches 395->397, 449->451, 468->470 in save_google_drive_settings. + """Cover branches 395->397, 449->451, 468->470 in save_google_drive_settings.""" - Note: the Google Drive save endpoint is named save_google_drive_settings in the - source (app/api/google_drive.py). - """ + @pytest.fixture(autouse=True) + def _admin_override(self): + from app.api.google_drive import _require_admin + from app.main import app as fastapi_app + + fastapi_app.dependency_overrides[_require_admin] = lambda: {"is_admin": True} + yield + fastapi_app.dependency_overrides.pop(_require_admin, None) @patch("app.api.google_drive.settings") @patch("os.path.exists", return_value=False) @@ -177,6 +182,7 @@ class TestSaveGoogleDriveSettingsFalsyFields: with patch("app.api.google_drive.notify_settings_updated"): result = await save_google_drive_settings( request=mock_request, + _admin={"is_admin": True}, refresh_token="", # falsy → branches 395->397 and 449->451 client_id="cid", client_secret=None, diff --git a/tests/test_wizard_db_persist.py b/tests/test_wizard_db_persist.py index 37197eaf..566ad01f 100644 --- a/tests/test_wizard_db_persist.py +++ b/tests/test_wizard_db_persist.py @@ -129,6 +129,15 @@ class TestSetupWizardUndoSkip: class TestDropboxSaveSettingsDbPersist: """Unit tests for save_dropbox_settings DB persistence.""" + @pytest.fixture(autouse=True) + def _admin_override(self): + from app.api.dropbox import _require_admin + from app.main import app as fastapi_app + + fastapi_app.dependency_overrides[_require_admin] = lambda: {"is_admin": True} + yield + fastapi_app.dependency_overrides.pop(_require_admin, None) + @patch("app.api.dropbox.settings") @patch("app.api.dropbox.notify_settings_updated") @patch("app.api.dropbox.save_setting_to_db") @@ -268,6 +277,15 @@ class TestGoogleDriveUpdateSettingsDbPersist: class TestOneDriveSaveSettingsDbPersist: """Unit tests for save_onedrive_settings DB persistence.""" + @pytest.fixture(autouse=True) + def _admin_override(self): + from app.api.onedrive import _require_admin + from app.main import app as fastapi_app + + fastapi_app.dependency_overrides[_require_admin] = lambda: {"is_admin": True} + yield + fastapi_app.dependency_overrides.pop(_require_admin, None) + @patch("app.api.onedrive.settings") @patch("app.api.onedrive.notify_settings_updated") @patch("app.api.onedrive.save_setting_to_db")