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

"""门户跳转登录接口：验签/按需落库 + 首次设密。"""

from __future__ import annotations

from flask import request

from dms.api.v1.blueprint import api_v1
from dms.common.errors import InvalidArgumentError, LoginIpForbiddenError
from dms.common.response import success_response
from dms.services.audit_service import record_login_failure
from dms.services.portal_auth_service import setup_portal_password, verify_portal_launch


def _verify_payload() -> str:
    if not request.is_json:
        raise InvalidArgumentError("请求体必须是JSON对象")
    body = request.get_json(silent=True)
    if not isinstance(body, dict) or set(body) != {"query"}:
        raise InvalidArgumentError("请求体必须且只能包含query")
    query = body["query"]
    if not isinstance(query, str) or not query.strip():
        raise InvalidArgumentError("query不能为空")
    return query.strip()


@api_v1.post("/auth/portal/verify")
def portal_verify():
    """门户跳转验签：透传浏览器地址栏query串，返回跳转阶段与用户名。"""
    return success_response(verify_portal_launch(_verify_payload()))


def _setup_payload() -> tuple[str, str, bool]:
    if not request.is_json:
        raise InvalidArgumentError("请求体必须是JSON对象")
    body = request.get_json(silent=True)
    if not isinstance(body, dict):
        raise InvalidArgumentError("请求体必须是JSON对象")
    expected = {"portalTicket", "newPassword", "keepSignedIn"}
    if set(body) != expected:
        raise InvalidArgumentError(
            "请求体必须且只能包含portalTicket、newPassword和keepSignedIn"
        )
    ticket = body["portalTicket"]
    password = body["newPassword"]
    keep_signed_in = body["keepSignedIn"]
    if not isinstance(ticket, str) or not ticket.strip():
        raise InvalidArgumentError("portalTicket不能为空")
    if not isinstance(password, str) or not password:
        raise InvalidArgumentError("newPassword不能为空")
    if not isinstance(keep_signed_in, bool):
        raise InvalidArgumentError("keepSignedIn必须是布尔值")
    return ticket.strip(), password, keep_signed_in


@api_v1.post("/auth/portal/setup-password")
def portal_setup_password():
    """门户用户首次设置独立密码并登录（票据一次性，5分钟内有效）。"""
    ticket, password, keep_signed_in = _setup_payload()
    try:
        payload = setup_portal_password(ticket, password, keep_signed_in)
    except LoginIpForbiddenError:
        record_login_failure(
            user=None,
            attempted_username="(portal-setup)",
            failure_reason="IP_NOT_ALLOWED",
        )
        raise
    return success_response(payload)