document-management/backend/dms/api/v1/auth.py

"""登录与注销接口。"""

from __future__ import annotations

from datetime import datetime, timezone

from flask import current_app, request
from sqlalchemy import select

from dms.api.v1.blueprint import api_v1
from dms.common.enums import AuditAction, AuditResult, EnabledStatus
from dms.common.errors import (
    AccountLockedError,
    InvalidArgumentError,
    InvalidCredentialsError,
    LoginIpForbiddenError,
    UserDisabledError,
)
from dms.services.captcha_service import issue as captcha_issue, verify as captcha_verify
from dms.common.response import success_response
from dms.database.transaction import for_update, transaction
from dms.extensions import db
from dms.models import User
from dms.security.auth_context import get_auth_context
from dms.security.decorators import bearer_auth_required
from dms.security.jwt_tokens import issue_access_token, issue_reauth_token
from dms.security.passwords import verify_password
from dms.services.admin_permission import SUPER_ADMIN_USERNAME
from dms.services.audit_service import auth_audit, client_ip, record_login_failure
from dms.services.ip_entry_service import verify_login_ip
from dms.services.login_lockout import (
    clear_failed_attempts,
    is_locked,
    register_failed_password_attempt,
)
from dms.services.user_service import allowed_modules_for_user, user_summary


def _login_payload() -> tuple[str, str, bool, str | None, str | None]:
    if not request.is_json:
        raise InvalidArgumentError("请求体必须是JSON对象")
    body = request.get_json(silent=True)
    if not isinstance(body, dict):
        raise InvalidArgumentError("请求体必须是JSON对象")
    # 验证码开启（安全需求4）时请求体扩为五键，关闭时保持三键不变
    captcha_enabled = current_app.config.get("DMS_CAPTCHA_ENABLED", True)
    expected = {"username", "password", "keepSignedIn"}
    if captcha_enabled:
        expected |= {"captchaId", "captchaCode"}
    if set(body) != expected:
        raise InvalidArgumentError(
            "请求体必须且只能包含username、password和keepSignedIn"
            + ("、captchaId和captchaCode" if captcha_enabled else "")
        )
    username = body["username"]
    password = body["password"]
    keep_signed_in = body["keepSignedIn"]
    if not isinstance(username, str) or not username.strip():
        raise InvalidArgumentError("username不能为空")
    if not isinstance(password, str) or not password:
        raise InvalidArgumentError("password不能为空")
    if not isinstance(keep_signed_in, bool):
        raise InvalidArgumentError("keepSignedIn必须是布尔值")
    captcha_id: str | None = None
    captcha_code: str | None = None
    if captcha_enabled:
        captcha_id = body["captchaId"]
        captcha_code = body["captchaCode"]
        if not isinstance(captcha_id, str) or not captcha_id.strip():
            raise InvalidArgumentError("captchaId不能为空")
        if not isinstance(captcha_code, str) or not captcha_code.strip():
            raise InvalidArgumentError("captchaCode不能为空")
    return username.strip(), password, keep_signed_in, captcha_id, captcha_code


@api_v1.get("/auth/captcha")
def captcha():
    """登录图片验证码（安全需求4）：开启时返回 data URL 图片，关闭时仅返回 enabled=false。"""
    if not current_app.config.get("DMS_CAPTCHA_ENABLED", True):
        return success_response({"enabled": False})
    data: dict[str, object] = captcha_issue()
    data["enabled"] = True
    return success_response(data)


@api_v1.post("/auth/login")
def login():
    username, password, keep_signed_in, captcha_id, captcha_code = _login_payload()
    # 验证码是防爆破第一道闸：先于用户查询/锁定检查，一次性消费不计锁定；
    # 开关关闭时直接放行（对齐 Java CaptchaService.verify，登录回归三键契约）
    if current_app.config.get("DMS_CAPTCHA_ENABLED", True):
        captcha_verify(captcha_id, captcha_code)
    # 超级管理员登录开关（安全需求7）关闭时按用户不存在处理：锁定/状态/IP
    # 检查全部短路，报错与密码错误完全一致，不泄露账号存在
    admin_login_blocked = (
        username == SUPER_ADMIN_USERNAME
        and not current_app.config.get("DMS_ALLOW_ADMIN_LOGIN", True)
    )
    user = (
        None
        if admin_login_blocked
        else db.session.scalar(
            select(User).where(User.username == username, User.is_deleted.is_(False))
        )
    )
    # 锁定检查先于密码校验：锁定期间一律拒绝，不延长锁定窗口
    if user is not None and is_locked(user):
        record_login_failure(
            user=user,
            attempted_username=username,
            failure_reason="ACCOUNT_LOCKED",
        )
        raise AccountLockedError()
    password_valid = verify_password(
        user.password_hash if user is not None else None,
        password,
    )
    if user is None or not password_valid:
        record_login_failure(
            user=user,
            attempted_username=username,
            failure_reason="INVALID_CREDENTIALS",
        )
        # 仅密码错误计入锁定计数（用户名不存在无账号可锁）
        if user is not None:
            register_failed_password_attempt(user)
        raise InvalidCredentialsError()
    if user.status != EnabledStatus.ENABLED.value:
        record_login_failure(
            user=user,
            attempted_username=username,
            failure_reason="USER_DISABLED",
        )
        raise UserDisabledError()

    try:
        verify_login_ip(user)
    except LoginIpForbiddenError:
        record_login_failure(
            user=user,
            attempted_username=username,
            failure_reason="IP_NOT_ALLOWED",
        )
        raise

    token, expires_in = issue_access_token(user, keep_signed_in=keep_signed_in)
    with transaction() as session:
        user.last_login_at = datetime.now(timezone.utc).replace(tzinfo=None)
        user.last_login_ip = client_ip()
        clear_failed_attempts(user)
        user.row_version += 1
        session.add(
            auth_audit(
                action=AuditAction.LOGIN,
                result=AuditResult.SUCCESS,
                user=user,
                detail={"keepSignedIn": keep_signed_in},
            )
        )

    return success_response(
        {
            "accessToken": token,
            "tokenType": "Bearer",
            "expiresIn": expires_in,
            # 页面空闲自动登出（契约1.29）：分钟数，0 表示关闭
            "idleLogoutMinutes": current_app.config["DMS_IDLE_LOGOUT_MINUTES"],
            "user": user_summary(
                user, allowed_modules=allowed_modules_for_user(user)
            ),
        }
    )


@api_v1.post("/auth/logout")
@bearer_auth_required
def logout():
    context = get_auth_context()
    with transaction() as session:
        user = session.scalar(for_update(select(User).where(User.id == context.user_id)))
        if user is None or user.is_deleted:
            raise InvalidCredentialsError()
        user.auth_version += 1
        user.row_version += 1
        session.add(
            auth_audit(
                action=AuditAction.LOGOUT,
                result=AuditResult.SUCCESS,
                user=user,
                detail={"authVersionIncremented": True},
            )
        )
    return success_response()


@api_v1.post("/auth/reauth")
@bearer_auth_required
def reauth():
    """敏感操作二次认证：登录态重输登录密码，换取 5 分钟 reauth 令牌（契约1.21）。"""
    if not request.is_json:
        raise InvalidArgumentError("请求体必须是JSON对象")
    body = request.get_json(silent=True)
    if not isinstance(body, dict) or set(body) != {"password"}:
        raise InvalidArgumentError("请求体必须且只能包含password")
    password = body["password"]
    if not isinstance(password, str) or not password:
        raise InvalidArgumentError("password不能为空")
    context = get_auth_context()
    user = db.session.scalar(
        select(User).where(User.id == context.user_id, User.is_deleted.is_(False))
    )
    if user is None or not verify_password(user.password_hash, password):
        raise InvalidCredentialsError()
    token, expires_in = issue_reauth_token(user)
    return success_response({"reauthToken": token, "expiresIn": expires_in})