Profeto/src/api/routes/auth.py

"""管理后台认证路由:密码登录 → HttpOnly Cookie 会话;支持在线修改密码。

管理员密码以 scrypt 哈希存于数据库(.env 明文仅作初始值,启动时自动迁移为哈希)。
修改密码会改变会话签名密钥,所有已登录会话随之失效,需重新登录。
"""
from __future__ import annotations

import logging
import secrets
import time
from collections import defaultdict, deque

from fastapi import APIRouter, Depends, HTTPException, Request, Response
from pydantic import BaseModel

from src.api.deps import (
    SESSION_COOKIE,
    auth_configured,
    create_session_token,
    get_session_secret,
    require_admin,
    verify_session_token,
)
from src.core import crypto
from src.core.config import settings
from src.core.runtime_config import (
    get_admin_password_hash,
    get_setting_origin,
    set_admin_password_hash,
    verify_admin_password,
)

logger = logging.getLogger(__name__)

router = APIRouter(prefix="/api/v1/auth", tags=["auth"])

# 简易防爆破:10 分钟窗口内同一 IP 连续失败 5 次即锁定 10 分钟(内存态,重启清零)
_MAX_FAILS = 5
_WINDOW_SECONDS = 600
_fail_times: dict[str, deque[float]] = defaultdict(deque)

# 新密码强度要求
_MIN_PASSWORD_LEN = 8
_MAX_PASSWORD_LEN = 128


class LoginIn(BaseModel):
    password: str


class PasswordChangeIn(BaseModel):
    current_password: str
    new_password: str


def _client_ip(request: Request) -> str:
    """获取客户端 IP,与限流共用同一套逻辑(防伪造)。"""
    from src.api.deps import get_client_ip
    return get_client_ip(request)


def _is_locked(ip: str) -> bool:
    dq = _fail_times.get(ip)
    if not dq:
        return False
    now = time.time()
    while dq and now - dq[0] > _WINDOW_SECONDS:
        dq.popleft()
    return len(dq) >= _MAX_FAILS


@router.post("/login")
async def login(body: LoginIn, request: Request, response: Response):
    ip = _client_ip(request)
    if not (await get_admin_password_hash() or settings.ADMIN_PASSWORD):
        raise HTTPException(status_code=503, detail="服务器未配置管理员密码,登录不可用")
    if _is_locked(ip):
        logger.warning("管理员登录尝试过于频繁 (ip=%s)", ip)
        raise HTTPException(status_code=429, detail="失败次数过多,请 10 分钟后再试")
    if not await verify_admin_password(body.password):
        _fail_times[ip].append(time.time())
        logger.warning("管理员登录失败 (ip=%s)", ip)
        raise HTTPException(status_code=401, detail="密码错误")

    _fail_times.pop(ip, None)
    response.set_cookie(
        key=SESSION_COOKIE,
        value=create_session_token(await get_session_secret()),
        max_age=settings.ADMIN_SESSION_TTL_HOURS * 3600,
        httponly=True,
        samesite="lax",
        path="/",
    )
    logger.info("管理员登录成功 (ip=%s)", ip)
    return {"ok": True, "expires_in_hours": settings.ADMIN_SESSION_TTL_HOURS}


@router.post("/logout")
async def logout(response: Response):
    response.delete_cookie(key=SESSION_COOKIE, path="/")
    return {"ok": True}


@router.get("/me")
async def me(request: Request):
    """前端登录门禁探测。未启用鉴权时视为已登录(本地开发模式)。"""
    token = request.cookies.get(SESSION_COOKIE)
    authenticated = not await auth_configured() or bool(
        token and verify_session_token(token, await get_session_secret())
    )
    has_hash = bool(await get_admin_password_hash())
    return {
        "authenticated": authenticated,
        "enabled": await auth_configured(),
        "password_origin": "db" if has_hash else ("env" if settings.ADMIN_PASSWORD else "none"),
    }


@router.post("/change-password", dependencies=[Depends(require_admin)])
async def change_password(body: PasswordChangeIn, request: Request, response: Response):
    """修改管理员密码:验证当前密码 → 写运行时覆盖 → 清除会话(全端登出)。"""
    if not await auth_configured():
        raise HTTPException(status_code=503, detail="服务器未配置管理员密码,无法修改")
    if not await verify_admin_password(body.current_password):
        logger.warning("修改密码失败:当前密码错误 (ip=%s)", _client_ip(request))
        raise HTTPException(status_code=401, detail="当前密码错误")

    new = body.new_password
    if not (_MIN_PASSWORD_LEN <= len(new) <= _MAX_PASSWORD_LEN):
        raise HTTPException(status_code=400, detail=f"新密码长度需在 {_MIN_PASSWORD_LEN}-{_MAX_PASSWORD_LEN} 位之间")
    if await verify_admin_password(new):
        raise HTTPException(status_code=400, detail="新密码不能与当前密码相同")

    await set_admin_password_hash(crypto.hash_password(new))
    # 密码即会话签名密钥,修改后所有旧会话失效;主动清除当前 Cookie 要求重新登录
    response.delete_cookie(key=SESSION_COOKIE, path="/")
    logger.info("管理员密码已修改 (ip=%s),所有会话已失效", _client_ip(request))
    return {"ok": True, "message": "密码已修改,请用新密码重新登录"}