85 lines
3.3 KiB
Python
85 lines
3.3 KiB
Python
"""认证 API"""
|
|
import jwt
|
|
import threading
|
|
import time
|
|
from datetime import datetime, timedelta, timezone
|
|
from fastapi import APIRouter, Depends, HTTPException, Request
|
|
from sqlalchemy.orm import Session
|
|
|
|
from ..schemas import LoginRequest, LoginResponse
|
|
from ..config import (
|
|
JWT_SECRET, JWT_ALGORITHM, JWT_EXPIRE_HOURS,
|
|
LOGIN_RATE_LIMIT_MAX_ATTEMPTS, LOGIN_RATE_LIMIT_WINDOW_SECONDS, LOGIN_RATE_LIMIT_LOCKOUT_SECONDS,
|
|
)
|
|
from ..deps import get_current_user
|
|
from ..database import get_db
|
|
from ..models import User
|
|
from ..security import hash_password, verify_password
|
|
|
|
router = APIRouter(prefix="/api/auth", tags=["auth"])
|
|
_login_attempts: dict[str, tuple[int, float, float]] = {}
|
|
_login_attempts_lock = threading.Lock()
|
|
|
|
|
|
def _create_token(username: str, is_admin: bool) -> str:
|
|
"""生成 JWT token"""
|
|
payload = {
|
|
"sub": username,
|
|
"is_admin": is_admin,
|
|
"exp": datetime.now(timezone.utc) + timedelta(hours=JWT_EXPIRE_HOURS),
|
|
}
|
|
return jwt.encode(payload, JWT_SECRET, algorithm=JWT_ALGORITHM)
|
|
|
|
|
|
def _check_login_rate_limit(client_ip: str) -> None:
|
|
"""按来源 IP 限制登录失败次数，降低在线暴力破解风险。"""
|
|
now = time.monotonic()
|
|
with _login_attempts_lock:
|
|
attempts, first_failure, locked_until = _login_attempts.get(client_ip, (0, now, 0))
|
|
if locked_until > now:
|
|
raise HTTPException(status_code=429, detail="登录失败次数过多，请稍后再试")
|
|
if now - first_failure > LOGIN_RATE_LIMIT_WINDOW_SECONDS:
|
|
_login_attempts.pop(client_ip, None)
|
|
|
|
|
|
def _record_login_failure(client_ip: str) -> None:
|
|
now = time.monotonic()
|
|
with _login_attempts_lock:
|
|
attempts, first_failure, _ = _login_attempts.get(client_ip, (0, now, 0))
|
|
if now - first_failure > LOGIN_RATE_LIMIT_WINDOW_SECONDS:
|
|
attempts, first_failure = 0, now
|
|
attempts += 1
|
|
locked_until = now + LOGIN_RATE_LIMIT_LOCKOUT_SECONDS if attempts >= LOGIN_RATE_LIMIT_MAX_ATTEMPTS else 0
|
|
_login_attempts[client_ip] = (attempts, first_failure, locked_until)
|
|
|
|
|
|
def _clear_login_failures(client_ip: str) -> None:
|
|
with _login_attempts_lock:
|
|
_login_attempts.pop(client_ip, None)
|
|
|
|
|
|
@router.post("/login", response_model=LoginResponse)
|
|
async def login(credentials: LoginRequest, request: Request, db: Session = Depends(get_db)):
|
|
"""用户登录"""
|
|
client_ip = request.client.host if request.client else "unknown"
|
|
_check_login_rate_limit(client_ip)
|
|
user = db.query(User).filter(User.username == credentials.username).first()
|
|
valid, needs_upgrade = verify_password(credentials.password, user.password_hash) if user else (False, False)
|
|
if not user or not valid:
|
|
_record_login_failure(client_ip)
|
|
raise HTTPException(status_code=401, detail="用户名或密码错误")
|
|
|
|
if needs_upgrade:
|
|
user.password_hash = hash_password(credentials.password)
|
|
db.commit()
|
|
_clear_login_failures(client_ip)
|
|
|
|
token = _create_token(user.username, user.is_admin)
|
|
return LoginResponse(token=token, username=user.username, is_admin=user.is_admin)
|
|
|
|
|
|
@router.get("/me")
|
|
async def get_me(user: dict = Depends(get_current_user)):
|
|
"""校验 token 有效性，返回当前用户信息"""
|
|
return {"username": user["username"], "is_admin": user["is_admin"]}
|