Kyle Pope
ab7e4a7c7e
PT-03: Make UMBRA_URL configurable via env var (default http://localhost). Replaces hardcoded http://10.0.69.35 in notification dispatch job and ntfy test endpoint. Add UMBRA_URL to .env.example. PT-05: Add explicit path="/" to session cookie for clarity. PT-06: Add concurrent session limit (MAX_SESSIONS_PER_USER, default 10). When exceeded, oldest sessions are revoked. New login always succeeds. PT-07: Escape LIKE metacharacters (%, _) in admin audit log action filter to prevent wildcard abuse. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
122 lines
4.4 KiB
Python
122 lines
4.4 KiB
Python
from fastapi import APIRouter, Depends, HTTPException
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.database import get_db
|
|
from app.config import settings as app_settings
|
|
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,
|
|
user_id=s.user_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
|
|
auto_lock_enabled=s.auto_lock_enabled,
|
|
auto_lock_minutes=s.auto_lock_minutes,
|
|
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=app_settings.UMBRA_URL,
|
|
)
|
|
|
|
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"}
|