Add login lockout with ntfy alerts and update docs
- Lock accounts for 15 minutes after 5 consecutive failed login attempts - Send urgent ntfy notification when an account is locked - Send high-priority ntfy notification on login attempt against a locked account - Auto-reset lockout on expiry; reset counter on successful login - Add v2.4 migration for failed_login_attempts and locked_until columns - Add ALLOWED_ORIGINS and SECURE_COOKIES to .env.example - Update README: lockout row in security table, new ntfy events Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
@@ -1,3 +1,5 @@
|
||||
import math
|
||||
from datetime import datetime, timezone, timedelta
|
||||
from zoneinfo import ZoneInfo, ZoneInfoNotFoundError
|
||||
|
||||
from fastapi import APIRouter, BackgroundTasks, Depends, HTTPException, Request, Response, status
|
||||
@@ -15,6 +17,9 @@ from auth import (
|
||||
|
||||
router = APIRouter(prefix="/api/auth", tags=["auth"])
|
||||
|
||||
_LOGIN_MAX_ATTEMPTS = 5
|
||||
_LOGIN_LOCKOUT_MINUTES = 15
|
||||
|
||||
|
||||
def _issue(response: Response, user: User, admin_id=None) -> AuthResponse:
|
||||
token = create_access_token(user.id, user.username, user.is_admin, user.timezone, admin_id=admin_id)
|
||||
@@ -24,20 +29,68 @@ def _issue(response: Response, user: User, admin_id=None) -> AuthResponse:
|
||||
|
||||
@router.post("/login", response_model=AuthResponse)
|
||||
def login(body: LoginRequest, response: Response, request: Request, background_tasks: BackgroundTasks, db: Session = Depends(get_db)):
|
||||
ip = request.headers.get("X-Forwarded-For", request.client.host if request.client else "unknown")
|
||||
ua = request.headers.get("User-Agent", "unknown")
|
||||
|
||||
user = db.scalars(select(User).where(User.username == body.username)).first()
|
||||
|
||||
# Check lockout before verifying password (don't reveal whether account exists)
|
||||
if user and user.locked_until:
|
||||
now = datetime.now(timezone.utc).replace(tzinfo=None)
|
||||
if user.locked_until > now:
|
||||
remaining = math.ceil((user.locked_until - now).total_seconds() / 60)
|
||||
background_tasks.add_task(
|
||||
notify,
|
||||
title="Yolkbook Login Attempt on Locked Account",
|
||||
message=f"User: {user.username}\nLocked for {remaining} more minute(s)\nIP: {ip}\nUA: {ua}",
|
||||
priority="high",
|
||||
tags=["lock"],
|
||||
)
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_403_FORBIDDEN,
|
||||
detail=f"Account is temporarily locked. Try again in {remaining} minute(s).",
|
||||
)
|
||||
else:
|
||||
# Lockout expired — reset counters
|
||||
user.failed_login_attempts = 0
|
||||
user.locked_until = None
|
||||
db.commit()
|
||||
|
||||
if not user or not verify_password(body.password, user.hashed_password):
|
||||
if user:
|
||||
user.failed_login_attempts += 1
|
||||
if user.failed_login_attempts >= _LOGIN_MAX_ATTEMPTS:
|
||||
user.locked_until = datetime.now(timezone.utc).replace(tzinfo=None) + timedelta(minutes=_LOGIN_LOCKOUT_MINUTES)
|
||||
db.commit()
|
||||
background_tasks.add_task(
|
||||
notify,
|
||||
title="Yolkbook Account Locked",
|
||||
message=f"User: {user.username}\nLocked after {user.failed_login_attempts} failed attempts\nIP: {ip}\nUA: {ua}",
|
||||
priority="urgent",
|
||||
tags=["warning"],
|
||||
)
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_403_FORBIDDEN,
|
||||
detail=f"Account locked after too many failed attempts. Try again in {_LOGIN_LOCKOUT_MINUTES} minutes.",
|
||||
)
|
||||
db.commit()
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_401_UNAUTHORIZED,
|
||||
detail="Invalid username or password",
|
||||
)
|
||||
|
||||
if user.is_disabled:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_403_FORBIDDEN,
|
||||
detail="Account is disabled. Contact your administrator.",
|
||||
)
|
||||
|
||||
# Successful login — reset failure counters
|
||||
user.failed_login_attempts = 0
|
||||
user.locked_until = None
|
||||
db.commit()
|
||||
|
||||
if user.is_admin:
|
||||
ip = request.headers.get("X-Forwarded-For", request.client.host if request.client else "unknown")
|
||||
ua = request.headers.get("User-Agent", "unknown")
|
||||
background_tasks.add_task(
|
||||
notify,
|
||||
title="Yolkbook Admin Login",
|
||||
|
||||
Reference in New Issue
Block a user