document-management/backend/dms/services/portal_auth_service.py

"""门户跳转登录服务：验签 → 用户匹配/按需落库 → 独立密码登录或首次设密。

流程（见 docs/门户对接指南.md）：
1. verify：提取token → 对端验签 → external_id/username 匹配本地用户；
   不存在则落库（密码为随机占位，不可登录），存在则回填external_id并同步展示信息；
   用户从未登录过（last_login_at为空）→ SET_PASSWORD（发一次性票据），否则 LOGIN。
2. setup-password：凭票据设置独立密码并完成登录（复用IP限制/审计/JWT签发）。
   设密成功后 auth_version+1，票据随之失效（天然一次性）。
"""

from __future__ import annotations

import secrets
from datetime import datetime, timedelta, timezone

import jwt
from flask import current_app
from sqlalchemy import select

from dms.common.enums import AuditAction, AuditResult, EnabledStatus
from dms.common.errors import (
    InternalError,
    InvalidArgumentError,
    PortalTicketInvalidError,
    PortalTokenInvalidError,
    UserDisabledError,
)
from dms.database.transaction import for_update, transaction
from dms.models import Organization, User
from dms.security.jwt_tokens import issue_access_token
from dms.security.passwords import hash_password
from dms.services.audit_service import auth_audit, client_ip
from dms.services.ip_entry_service import verify_login_ip
from dms.services.portal_config import (
    PortalConfigError,
    load_portal_config,
)
from dms.services.portal_provider import (
    PortalIdentityError,
    PortalTokenError,
    extract_token,
    resolve_portal_identity,
)
from dms.services.user_service import user_summary

SETUP_TICKET_TTL_SECONDS = 300
_TICKET_PURPOSE = "PORTAL_SETUP"


def _validate_portal_password(value: str) -> None:
    # 与 user_service 的密码规则保持一致（8~64位）
    if not isinstance(value, str) or not value:
        raise InvalidArgumentError("密码不能为空")
    if len(value) < 8 or len(value) > 64:
        raise InvalidArgumentError("密码长度必须在8~64个字符之间")


def _match_organization(
    session, org_name: str | None
) -> tuple[int | None, str | None]:
    """按组织名称精确匹配（未删）；匹配不到回落默认组织，再不行不挂组织。"""
    if org_name:
        org = session.scalar(
            select(Organization).where(
                Organization.org_name == org_name,
                Organization.is_deleted.is_(False),
            )
        )
        if org is not None:
            return org.id, org.org_name
    default_org_id = load_portal_config().default_org_id
    if default_org_id:
        org = session.get(Organization, default_org_id)
        if org is not None and not org.is_deleted:
            return org.id, org.org_name
    return None, None


def verify_portal_launch(query_string: str) -> dict[str, object]:
    """门户跳转验签入口，返回 phase/username/realName（及SET_PASSWORD时的票据）。"""
    if not load_portal_config().enabled:
        raise InvalidArgumentError("门户登录未启用")
    token = extract_token(query_string)
    if token is None:
        raise InvalidArgumentError(
            "跳转链接缺少token参数（参数名以DMS_PORTAL_TOKEN_PARAM配置为准）"
        )

    try:
        identity = resolve_portal_identity(token)
    except PortalConfigError as exc:
        raise InternalError(f"门户配置错误：{exc}") from exc
    except PortalTokenError as exc:
        raise PortalTokenInvalidError(str(exc)) from exc
    except PortalIdentityError as exc:
        raise InternalError(f"对端用户信息不完整：{exc}") from exc

    if len(identity.username) > 64 or len(identity.real_name) > 64:
        raise InternalError("对端用户名或姓名超出本系统长度限制（64字符）")

    with transaction() as session:
        user = session.scalar(
            select(User).where(
                User.external_id == identity.external_id,
                User.is_deleted.is_(False),
            )
        )
        if user is None:
            user = session.scalar(
                select(User).where(
                    User.username == identity.username,
                    User.is_deleted.is_(False),
                )
            )

        org_id, org_name = _match_organization(session, identity.org_name)
        detail = {
            "portal": True,
            "externalId": identity.external_id,
        }

        if user is None:
            user = User(
                username=identity.username,
                # 随机占位密码：在设置独立密码前任何密码都无法通过校验
                password_hash=hash_password(secrets.token_urlsafe(32)),
                real_name=identity.real_name,
                external_id=identity.external_id,
                organization_id=org_id,
                organization_name=org_name,
                role_code="USER",
                security_level="INTERNAL",
                status="ENABLED",
            )
            session.add(user)
            session.flush()
            session.add(
                auth_audit(
                    action=AuditAction.CREATE_USER,
                    result=AuditResult.SUCCESS,
                    user=user,
                    detail=detail,
                )
            )
        else:
            changed = False
            if user.external_id != identity.external_id:
                user.external_id = identity.external_id
                changed = True
            # username 不跟随对端变更（避免与本地其他用户冲突），以本地为准登录
            if identity.real_name and user.real_name != identity.real_name:
                user.real_name = identity.real_name
                changed = True
            if org_id is not None and user.organization_id != org_id:
                user.organization_id = org_id
                user.organization_name = org_name
                changed = True
            if changed:
                user.row_version += 1
                session.add(
                    auth_audit(
                        action=AuditAction.EDIT_USER,
                        result=AuditResult.SUCCESS,
                        user=user,
                        detail=detail,
                    )
                )

        if user.status != EnabledStatus.ENABLED.value:
            raise UserDisabledError()

        if user.last_login_at is None:
            return {
                "phase": "SET_PASSWORD",
                "username": user.username,
                "realName": user.real_name,
                "portalTicket": _issue_setup_ticket(user),
            }
        return {
            "phase": "LOGIN",
            "username": user.username,
            "realName": user.real_name,
        }


def _ticket_secret() -> str:
    secret = str(current_app.config.get("DMS_JWT_SECRET", "")).strip()
    if not secret:
        raise InternalError("JWT签名密钥未配置")
    return secret


def _issue_setup_ticket(user: User) -> str:
    now = datetime.now(timezone.utc)
    claims = {
        "sub": str(user.id),
        "purpose": _TICKET_PURPOSE,
        "authVersion": user.auth_version,
        "iat": now,
        "exp": now + timedelta(seconds=SETUP_TICKET_TTL_SECONDS),
        "iss": current_app.config["DMS_JWT_ISSUER"],
    }
    return jwt.encode(claims, _ticket_secret(), algorithm="HS256")


def _consume_setup_ticket(session, ticket: str) -> User:
    """校验并消费票据：过期/复用（auth_version已变）/用户无效均拒绝。"""
    try:
        claims = jwt.decode(
            ticket,
            _ticket_secret(),
            algorithms=["HS256"],
            issuer=current_app.config["DMS_JWT_ISSUER"],
            options={"require": ["sub", "purpose", "authVersion", "exp"]},
        )
    except jwt.PyJWTError as exc:
        raise PortalTicketInvalidError() from exc
    if claims.get("purpose") != _TICKET_PURPOSE or not str(claims.get("sub", "")).isdigit():
        raise PortalTicketInvalidError()
    user = session.scalar(
        for_update(select(User).where(User.id == int(claims["sub"])))
    )
    if user is None or user.is_deleted:
        raise PortalTicketInvalidError()
    if user.auth_version != claims.get("authVersion"):
        # 票据已使用过（设密后auth_version+1）或用户状态已变更
        raise PortalTicketInvalidError("设置密码凭证已使用，请从门户重新跳转")
    return user


def setup_portal_password(
    ticket: str, new_password: str, keep_signed_in: bool
) -> dict[str, object]:
    """凭票据设置独立密码并直接完成登录（返回与/auth/login一致的结构）。"""
    _validate_portal_password(new_password)
    with transaction() as session:
        user = _consume_setup_ticket(session, ticket)
        if user.status != EnabledStatus.ENABLED.value:
            raise UserDisabledError()
        # IP登录限制与普通登录一致（ADMIN豁免、无部门/无分配放行）
        verify_login_ip(user)
        user.password_hash = hash_password(new_password)
        user.auth_version += 1  # 使票据一次性失效
        user.last_login_at = datetime.now(timezone.utc).replace(tzinfo=None)
        user.last_login_ip = client_ip()
        user.row_version += 1
        session.add(
            auth_audit(
                action=AuditAction.LOGIN,
                result=AuditResult.SUCCESS,
                user=user,
                detail={"portal": True, "setupPassword": True},
            )
        )
        token, expires_in = issue_access_token(user, keep_signed_in=keep_signed_in)
    return {
        "accessToken": token,
        "tokenType": "Bearer",
        "expiresIn": expires_in,
        "user": user_summary(user),
    }