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, str | None]:
    if not request.is_json:
        raise InvalidArgumentError("请求体必须是JSON对象")
    body = request.get_json(silent=True)
    if not isinstance(body, dict):
        raise InvalidArgumentError("请求体必须是JSON对象")
    required = {"portalTicket", "newPassword", "keepSignedIn"}
    allowed = required | {"organizationId"}
    if not required.issubset(set(body)) or set(body) - allowed:
        raise InvalidArgumentError(
            "请求体必须包含portalTicket、newPassword、keepSignedIn"
            + "（organizationId可选）"
        )
    ticket = body["portalTicket"]
    password = body["newPassword"]
    keep_signed_in = body["keepSignedIn"]
    organization_id = body.get("organizationId")
    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必须是布尔值")
    if organization_id is not None and (
        not isinstance(organization_id, str) or not organization_id.strip()
    ):
        raise InvalidArgumentError("organizationId必须是非空字符串")
    return (
        ticket.strip(),
        password,
        keep_signed_in,
        organization_id.strip() if organization_id is not None else None,
    )


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