from datetime import datetime, timedelta
|
|
from fastapi import HTTPException
|
from sqlalchemy.orm import Session
|
|
from app.config import settings
|
from app.models import LoginLock
|
|
|
def _lock_message(locked_until: datetime) -> str:
|
remain = int((locked_until - datetime.utcnow()).total_seconds())
|
if remain < 60:
|
return f"登录尝试过多，请 {max(remain, 1)} 秒后重试"
|
minutes = max(remain // 60, 1)
|
return f"登录尝试过多，请 {minutes} 分钟后重试"
|
|
|
def ensure_login_allowed(db: Session, username: str) -> None:
|
key = username.strip().lower()
|
if not key:
|
return
|
row = db.get(LoginLock, key)
|
if not row or not row.locked_until:
|
return
|
if row.locked_until <= datetime.utcnow():
|
row.fail_count = 0
|
row.locked_until = None
|
db.commit()
|
return
|
raise HTTPException(429, detail=_lock_message(row.locked_until))
|
|
|
def record_login_failure(db: Session, username: str) -> None:
|
key = username.strip().lower()
|
if not key:
|
return
|
row = db.get(LoginLock, key)
|
if not row:
|
row = LoginLock(username=key, fail_count=0)
|
db.add(row)
|
row.fail_count += 1
|
row.updated_at = datetime.utcnow()
|
if row.fail_count >= settings.login_max_failures:
|
row.locked_until = datetime.utcnow() + timedelta(minutes=settings.login_lock_minutes)
|
db.commit()
|
|
|
def clear_login_lock(db: Session, username: str) -> None:
|
key = username.strip().lower()
|
if not key:
|
return
|
row = db.get(LoginLock, key)
|
if row:
|
db.delete(row)
|
db.commit()
|
