Add ntfy push notifications for security-relevant events

Sends alerts on admin login, new registrations, user disable/delete, and
impersonation. NTFY_URL and NTFY_TOKEN are optional — leave blank to disable.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
2026-03-23 23:01:13 -07:00
parent 4172b63dc2
commit 7cd2dfb710
7 changed files with 108 additions and 4 deletions

View File

@@ -1,11 +1,12 @@
import logging
from fastapi import APIRouter, Depends, HTTPException, Response
from fastapi import APIRouter, BackgroundTasks, Depends, HTTPException, Request, Response
from sqlalchemy import select
from sqlalchemy.orm import Session
from database import get_db
from models import User
from ntfy import notify
from schemas import UserCreate, UserOut, ResetPasswordRequest, AuthResponse
from auth import hash_password, create_access_token, get_current_admin, get_token_payload, set_auth_cookie, token_to_user_payload
@@ -60,6 +61,8 @@ def reset_password(
@router.post("/users/{user_id}/disable")
def disable_user(
user_id: int,
request: Request,
background_tasks: BackgroundTasks,
current_admin: User = Depends(get_current_admin),
db: Session = Depends(get_db),
):
@@ -71,6 +74,14 @@ def disable_user(
user.is_disabled = True
db.commit()
logger.warning("Admin '%s' disabled user '%s' (id=%d).", current_admin.username, user.username, user.id)
ip = request.headers.get("X-Forwarded-For", request.client.host if request.client else "unknown")
background_tasks.add_task(
notify,
title="Yolkbook User Disabled",
message=f"User: {user.username}\nBy admin: {current_admin.username}\nIP: {ip}",
priority="high",
tags=["warning"],
)
return {"detail": f"User {user.username} disabled"}
@@ -91,6 +102,8 @@ def enable_user(
@router.delete("/users/{user_id}", status_code=204)
def delete_user(
user_id: int,
request: Request,
background_tasks: BackgroundTasks,
current_admin: User = Depends(get_current_admin),
db: Session = Depends(get_db),
):
@@ -100,6 +113,14 @@ def delete_user(
if user.id == current_admin.id:
raise HTTPException(status_code=400, detail="Cannot delete your own account")
logger.warning("Admin '%s' deleted user '%s' (id=%d).", current_admin.username, user.username, user.id)
ip = request.headers.get("X-Forwarded-For", request.client.host if request.client else "unknown")
background_tasks.add_task(
notify,
title="Yolkbook User Deleted",
message=f"User: {user.username}\nBy admin: {current_admin.username}\nIP: {ip}",
priority="urgent",
tags=["warning", "wastebasket"],
)
db.delete(user)
db.commit()
@@ -108,6 +129,8 @@ def delete_user(
def impersonate_user(
user_id: int,
response: Response,
request: Request,
background_tasks: BackgroundTasks,
current_admin: User = Depends(get_current_admin),
db: Session = Depends(get_db),
):
@@ -117,6 +140,14 @@ def impersonate_user(
token = create_access_token(user.id, user.username, user.is_admin, user.timezone, admin_id=current_admin.id)
set_auth_cookie(response, token)
logger.warning("Admin '%s' (id=%d) is impersonating user '%s' (id=%d).", current_admin.username, current_admin.id, user.username, user.id)
ip = request.headers.get("X-Forwarded-For", request.client.host if request.client else "unknown")
background_tasks.add_task(
notify,
title="Yolkbook Admin Impersonation",
message=f"Admin: {current_admin.username} → User: {user.username}\nIP: {ip}",
priority="high",
tags=["eyes"],
)
return AuthResponse(user=token_to_user_payload(token))