from fastapi import APIRouter, Depends, HTTPException from sqlalchemy.ext.asyncio import AsyncSession from app.database import get_db from app.models.settings import Settings from app.models.user import User from app.schemas.settings import SettingsUpdate, SettingsResponse from app.routers.auth import get_current_user, get_current_settings router = APIRouter() def _to_settings_response(s: Settings) -> SettingsResponse: """ Explicitly construct SettingsResponse, computing ntfy_has_token from the stored token without ever exposing the token value. """ return SettingsResponse( id=s.id, accent_color=s.accent_color, upcoming_days=s.upcoming_days, preferred_name=s.preferred_name, weather_city=s.weather_city, weather_lat=s.weather_lat, weather_lon=s.weather_lon, first_day_of_week=s.first_day_of_week, ntfy_server_url=s.ntfy_server_url, ntfy_topic=s.ntfy_topic, ntfy_enabled=s.ntfy_enabled, ntfy_events_enabled=s.ntfy_events_enabled, ntfy_reminders_enabled=s.ntfy_reminders_enabled, ntfy_todos_enabled=s.ntfy_todos_enabled, ntfy_projects_enabled=s.ntfy_projects_enabled, ntfy_event_lead_minutes=s.ntfy_event_lead_minutes, ntfy_todo_lead_days=s.ntfy_todo_lead_days, ntfy_project_lead_days=s.ntfy_project_lead_days, ntfy_has_token=bool(s.ntfy_auth_token), # derived — never expose the token value created_at=s.created_at, updated_at=s.updated_at, ) @router.get("/", response_model=SettingsResponse) async def get_settings( db: AsyncSession = Depends(get_db), current_settings: Settings = Depends(get_current_settings) ): """Get current settings (excluding ntfy auth token).""" return _to_settings_response(current_settings) @router.put("/", response_model=SettingsResponse) async def update_settings( settings_update: SettingsUpdate, db: AsyncSession = Depends(get_db), current_settings: Settings = Depends(get_current_settings) ): """Update settings.""" update_data = settings_update.model_dump(exclude_unset=True) for key, value in update_data.items(): setattr(current_settings, key, value) await db.commit() await db.refresh(current_settings) return _to_settings_response(current_settings) @router.post("/ntfy/test") async def test_ntfy( db: AsyncSession = Depends(get_db), current_settings: Settings = Depends(get_current_settings) ): """ Send a test ntfy notification to verify the user's configuration. Requires ntfy_server_url and ntfy_topic to be set. Note: ntfy_enabled does not need to be True to run the test — the service call bypasses that check because we pass settings directly. """ if not current_settings.ntfy_server_url or not current_settings.ntfy_topic: raise HTTPException( status_code=400, detail="ntfy server URL and topic must be configured before sending a test" ) # SSRF-validate the URL before attempting the outbound request from app.services.ntfy import validate_ntfy_host, send_ntfy_notification try: validate_ntfy_host(current_settings.ntfy_server_url) except ValueError as e: raise HTTPException(status_code=400, detail=str(e)) # Temporarily treat ntfy as enabled for the test send even if master switch is off class _TestSettings: """Thin wrapper that forces ntfy_enabled=True for the test call.""" ntfy_enabled = True ntfy_server_url = current_settings.ntfy_server_url ntfy_topic = current_settings.ntfy_topic ntfy_auth_token = current_settings.ntfy_auth_token success = await send_ntfy_notification( settings=_TestSettings(), # type: ignore[arg-type] title="UMBRA Test Notification", message="If you see this, your ntfy integration is working correctly.", tags=["white_check_mark"], priority=3, click_url="http://10.0.69.35", ) if not success: raise HTTPException( status_code=502, detail="Failed to deliver test notification — check server URL, topic, and auth token" ) return {"message": "Test notification sent successfully"}