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

"""方案密码锁：判定、验证与设置/修改/清除。

核心语义（需求已确认）：
- 一个 Argon2id 哈希密码 + 7 个操作勾选位，仅主案（MAIN）可设密；
- 普通用户：仅勾选操作需要密码；
- 管理员（ADMIN）：对设密主案的除 DELETE 外全部操作均需密码（防止越权查看），
  DELETE 免密但前端要求二次确认；
- 修改/清除密码时任何人（含管理员）都必须提供原密码；
- 密码经请求头 X-DMS-Document-Password 传递，避免进入URL/访问日志。
"""

from __future__ import annotations

from typing import Any

from flask import request
from sqlalchemy import select

from dms.common.enums import (
    AllowedAction,
    AuditAction,
    AuditTarget,
    DocumentPasswordAction,
    DocumentType,
    RoleCode,
)
from dms.common.errors import (
    ConflictError,
    DocumentPasswordInvalidError,
    DocumentPasswordRequiredError,
    InvalidArgumentError,
    ResourceNotFoundError,
)
from dms.common.response import serialize_id
from dms.database.transaction import transaction
from dms.models import LOCK_DEFER_LOB, Document
from dms.security.auth_context import AuthContext
from dms.security.passwords import hash_password, verify_password
from dms.services.audit_service import business_audit
from dms.services.authorization_service import require_plan_action

PASSWORD_HEADER = "X-DMS-Document-Password"
PASSWORD_FIELDS = {"password", "protectedActions"}
PASSWORD_MAX_LENGTH = 128

# 操作码 → 模型勾选位列名
_ACTION_COLUMNS: dict[str, str] = {
    DocumentPasswordAction.VIEW.value: "protect_view",
    DocumentPasswordAction.DOWNLOAD.value: "protect_download",
    DocumentPasswordAction.EDIT.value: "protect_edit",
    DocumentPasswordAction.DELETE.value: "protect_delete",
    DocumentPasswordAction.CONFIG_PERMISSION.value: (
        "protect_config_permission"
    ),
    DocumentPasswordAction.BIND_ATTACHMENT.value: "protect_bind_attachment",
    DocumentPasswordAction.UNBIND_ATTACHMENT.value: (
        "protect_unbind_attachment"
    ),
}


def parse_protected_actions(value: Any) -> list[str]:
    """校验并返回合法的保护操作码列表（按枚举定义顺序稳定排序）。"""
    if not isinstance(value, list) or not value:
        raise InvalidArgumentError(
            "protectedActions必须是非空数组，元素为"
            "VIEW/DOWNLOAD/EDIT/DELETE/CONFIG_PERMISSION/"
            "BIND_ATTACHMENT/UNBIND_ATTACHMENT"
        )
    valid = {action.value for action in DocumentPasswordAction}
    seen: list[str] = []
    for item in value:
        if not isinstance(item, str) or item not in valid:
            raise InvalidArgumentError(
                "protectedActions包含未知操作码：" + repr(item)
            )
        if item not in seen:
            seen.append(item)
    return [action.value for action in DocumentPasswordAction if action.value in seen]


def protected_actions_of(document: Document) -> list[str]:
    """文档当前勾选的保护操作码（未设密恒为空）。"""
    if document.password_hash is None:
        return []
    return [
        action
        for action, column in _ACTION_COLUMNS.items()
        if getattr(document, column)
    ]


def password_required_for(
    document: Document,
    action: str,
    context: AuthContext,
) -> bool:
    """该用户对该文档的指定操作是否需要密码。"""
    if document.password_hash is None:
        return False
    if context.role_code == RoleCode.ADMIN:
        # 管理员对设密主案除删除外全部操作均需密码（不限于勾选位）
        return action != DocumentPasswordAction.DELETE.value
    column = _ACTION_COLUMNS.get(action)
    if column is None:
        return False
    return bool(getattr(document, column))


def require_document_password(
    document: Document,
    action: str,
    password: str | None,
    context: AuthContext,
) -> None:
    """操作入口守卫：需要密码时校验，缺失/错误分别抛出对应错误。"""
    if not password_required_for(document, action, context):
        return
    if not password:
        raise DocumentPasswordRequiredError(
            details={"action": action, "documentId": serialize_id(document.id)}
        )
    if not verify_password(document.password_hash, password):
        raise DocumentPasswordInvalidError(
            details={"action": action, "documentId": serialize_id(document.id)}
        )


def get_request_password() -> str | None:
    value = request.headers.get(PASSWORD_HEADER)
    return value if value else None


def is_view_locked(document: Document, context: AuthContext) -> bool:
    """列表/详情序列化用：查看（文件正文）是否被密码锁拦截。"""
    return password_required_for(
        document, DocumentPasswordAction.VIEW.value, context
    )


def view_password_verified(document: Document, context: AuthContext) -> bool:
    """详情专用：查看需密码时请求头密码是否正确（不抛错，供字段屏蔽判定）。"""
    if not is_view_locked(document, context):
        return True
    password = get_request_password()
    return bool(password) and verify_password(document.password_hash, password)


def apply_password_lock(
    document: Document,
    password: str,
    actions: list[str],
) -> None:
    """写入密码哈希与勾选位（调用方保证仅 MAIN、已通过校验）。"""
    document.password_hash = hash_password(password)
    selected = set(actions)
    for action, column in _ACTION_COLUMNS.items():
        setattr(document, column, action in selected)


def clear_password_lock(document: Document) -> None:
    document.password_hash = None
    for column in _ACTION_COLUMNS.values():
        setattr(document, column, False)


def _pop_password_spec(
    payload: dict[str, Any],
    *,
    attachment: bool,
) -> tuple[str | None, list[str] | None]:
    """从上传 metadata 中摘出可选密码锁字段并校验（仅新增主案可用）。

    返回 (password, protectedActions)；未携带时为 (None, None)。
    """
    if not (PASSWORD_FIELDS & set(payload)):
        return None, None
    raw_type = payload.get("documentType")
    if attachment or raw_type != DocumentType.MAIN.value:
        raise InvalidArgumentError("密码锁仅支持上传主案（MAIN）时设置")
    if "overwriteDocumentId" in payload:
        raise InvalidArgumentError(
            "覆盖上传不能修改密码锁，请在上传后通过密码管理维护"
        )
    if "password" not in payload or "protectedActions" not in payload:
        raise InvalidArgumentError(
            "password与protectedActions必须同时提供"
        )
    password = payload.pop("password")
    if (
        not isinstance(password, str)
        or not password.strip()
        or len(password) > PASSWORD_MAX_LENGTH
    ):
        raise InvalidArgumentError(
            f"password必须是1~{PASSWORD_MAX_LENGTH}位的非空字符串"
        )
    actions = parse_protected_actions(payload.pop("protectedActions"))
    return password, actions


def _require_active_main_plan(session, document_id: int) -> Document:
    document = session.scalar(
        select(Document)
        .where(
            Document.id == document_id,
            Document.document_type == DocumentType.MAIN.value,
            Document.is_deleted.is_(False),
        )
        .options(*LOCK_DEFER_LOB)
        .with_for_update()
    )
    if document is None:
        raise ResourceNotFoundError("主案不存在或不可设置密码锁")
    return document


def _password_summary(document: Document) -> dict[str, object]:
    return {
        "id": serialize_id(document.id),
        "hasPassword": document.password_hash is not None,
        "protectedActions": protected_actions_of(document),
        "rowVersion": document.row_version,
    }


def save_document_password(document_id: int, payload: dict[str, Any]) -> dict[str, object]:
    """设置/修改/清除主案密码锁。

    payload: { oldPassword?, password?, protectedActions?, rowVersion }
    - 未设密时设置：password + protectedActions 必填，无需 oldPassword；
    - 已设密时修改/清除：oldPassword 必填且正确；password 为空/缺省即清除。
    """
    if not isinstance(payload, dict) or set(payload) - {
        "oldPassword",
        "password",
        "protectedActions",
        "rowVersion",
    }:
        raise InvalidArgumentError(
            "请求体只允许oldPassword、password、protectedActions、rowVersion"
        )
    raw_version = payload.get("rowVersion")
    if not isinstance(raw_version, int) or isinstance(raw_version, bool) or raw_version < 0:
        raise InvalidArgumentError("rowVersion必须是非负整数")
    row_version = raw_version
    old_password = payload.get("oldPassword")
    new_password = payload.get("password")
    with transaction() as session:
        document = _require_active_main_plan(session, document_id)
        # 密码锁是文档级安全配置：密级+授权位（CONFIG_PERMISSION）前置校验
        require_plan_action(
            document, get_auth_context(), AllowedAction.CONFIG_PERMISSION
        )
        if document.row_version != row_version:
            raise ConflictError(
                "方案已被其他操作修改，请刷新后重试",
                details={"currentRowVersion": document.row_version},
            )
        if document.password_hash is not None:
            if not old_password or not verify_password(
                document.password_hash, old_password
            ):
                raise DocumentPasswordInvalidError(
                    details={"reason": "oldPassword"}
                )
        elif old_password:
            raise InvalidArgumentError("该主案尚未设置密码，无需提供oldPassword")
        removing = not new_password
        if removing:
            if "protectedActions" in payload and payload["protectedActions"]:
                raise InvalidArgumentError(
                    "清除密码时不能同时指定protectedActions"
                )
            clear_password_lock(document)
            action = AuditAction.DOCUMENT_PASSWORD_REMOVE
            detail: dict[str, Any] = {}
        else:
            if (
                not isinstance(new_password, str)
                or not new_password.strip()
                or len(new_password) > PASSWORD_MAX_LENGTH
            ):
                raise InvalidArgumentError(
                    f"password必须是1~{PASSWORD_MAX_LENGTH}位的非空字符串"
                )
            actions = parse_protected_actions(payload.get("protectedActions"))
            apply_password_lock(document, new_password, actions)
            action = AuditAction.DOCUMENT_PASSWORD_SET
            detail = {"protectedActions": actions}
        document.row_version += 1
        detail["rowVersion"] = document.row_version
        session.add(
            business_audit(
                action=action,
                target=AuditTarget.DOCUMENT,
                target_id=document.id,
                target_name=document.document_name,
                detail=detail,
            )
        )
        result = _password_summary(document)
    return result