TanWenyan/leaudit-platform-backend
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
0f385c9839c4bb7c4883c8313ac15aa11efecfc9
leaudit-platform-backend/docs/权限与地区隔离/统一执行器落地代码骨架与接入示例.md
T

28 KiB
Raw Blame History

统一执行器落地代码骨架与接入示例

适用范围：leaudit-platform 后端权限改造实施阶段
文档定位：把“统一数据范围执行器设计”进一步落到代码骨架、接口签名、目录建议和模块接入示例层面，供研发直接照着实施。

1. 文档目标

前面几份文档已经回答了：

  • 为什么要做统一执行器
  • 哪些接口会受影响
  • 应该按什么阶段实施
  • 字段和 SQL 应该如何规范化

这一份只解决一个问题：

后端代码到底怎么改，才能在不推翻现有 service 结构的前提下，把统一执行器落进去。

核心要求：

  1. 尽量兼容现有 FastAPI + service 分层风格
  2. 优先通过“新增能力 + 渐进替换”推进
  3. 先不大改 controller 签名
  4. 先把 service 内部角色判断替换掉

2. 当前代码风格约束

从现有代码看，后端基本遵循：

  1. Controller
    • 负责接 JWT payload
    • 负责做基础 permission 判断
    • 调用 ServiceImpl
  2. Service
    • 负责业务逻辑
    • 直接写 SQL / 组装查询
  3. PermissionServiceImpl
    • 当前仅提供布尔权限判断

因此统一执行器落地时，不建议第一步就做成特别重的 AOP 或 decorator 体系。

更适合的落地方式是：

  1. 先保留 controller 现状
  2. 在 service 内引入统一 PermissionDecisionService
  3. 逐步把旧的 _getCurrentUserContext、_build...Filters 替换掉

3. 推荐目录结构

建议新增：

fastapi_modules/fastapi_leaudit/services/permission_scope/
  __init__.py
  models.py
  enums.py
  exceptions.py
  repositories/
    __init__.py
    permissionGrantRepository.py
    resourceScopeRepository.py
  resolvers/
    __init__.py
    dataScopeResolver.py
    permissionDecisionService.py
    queryScopeBuilder.py
  policies/
    __init__.py
    base.py
    documentPolicy.py
    govdocPolicy.py
    usageStatsPolicy.py
    ragPolicy.py
    crossReviewPolicy.py
    rbacAdminPolicy.py
    contractTemplatePolicy.py
  facade/
    __init__.py
    scopeAwarePermissionFacade.py

说明：

  • repositories 负责查授权和资源归属
  • resolvers 负责算决策
  • policies 负责模块特例
  • facade 负责给业务 service 暴露统一入口

4. 建议新增核心对象

4.1 models.py

建议定义：

from __future__ import annotations

from dataclasses import dataclass, field
from typing import Any


@dataclass(slots=True)
class ScopeContext:
    user_id: int
    permission_key: str
    module: str
    action: str
    user_area: str | None
    request_area: str | None = None
    target_user_id: int | None = None
    resource_id: int | None = None
    route_path: str | None = None
    extra: dict[str, Any] = field(default_factory=dict)


@dataclass(slots=True)
class PermissionGrant:
    role_id: int
    role_key: str
    permission_key: str
    grant_type: str
    role_scope: str | None
    permission_scope: str | None
    priority: int
    is_system_role: bool
    condition_filter: dict[str, Any] | None = None


@dataclass(slots=True)
class PermissionDecision:
    allowed: bool
    deny_reason: str | None
    effective_scope: str | None
    scope_mode: str
    allowed_areas: list[str] = field(default_factory=list)
    allowed_user_ids: list[int] = field(default_factory=list)
    allow_public: bool = False
    matched_roles: list[str] = field(default_factory=list)
    matched_permissions: list[str] = field(default_factory=list)
    conditions: dict[str, Any] = field(default_factory=dict)


@dataclass(slots=True)
class ScopeFieldMapping:
    area_field: str | None = None
    creator_field: str | None = None
    owner_field: str | None = None
    user_field: str | None = None
    public_field: str | None = None


@dataclass(slots=True)
class ScopeClause:
    sql: str
    params: dict[str, Any]
    scope_mode: str
    description: str

4.2 enums.py

建议定义：

class ScopeMode:
    NONE = "NONE"
    ALL = "ALL"
    DEPT = "DEPT"
    SELF = "SELF"
    RELATION = "RELATION"
    PUBLIC_MIXED = "PUBLIC_MIXED"
    CUSTOM = "CUSTOM"

5. 推荐新增权限服务能力

5.1 不直接推翻 IPermissionService

当前接口是：

  • CheckPermission
  • HasAnyPermission
  • HasAllPermissions

不建议直接改掉这 3 个方法，否则牵涉面太广。

建议做法：

  1. 保留原接口
  2. 额外新增一个 scope 感知服务接口

5.2 推荐新增接口

建议新增文件：

fastapi_modules/fastapi_leaudit/services/permissionScopeService.py

from abc import ABC, abstractmethod

from fastapi_modules.fastapi_leaudit.services.permission_scope.models import PermissionDecision, ScopeContext


class IPermissionScopeService(ABC):
    @abstractmethod
    async def Decide(self, Context: ScopeContext) -> PermissionDecision:
        ...

    @abstractmethod
    async def Require(self, Context: ScopeContext) -> PermissionDecision:
        ...

说明：

  • Decide() 返回决策，不抛错
  • Require() 返回决策，若拒绝直接抛 403

5.3 为什么不直接塞进 IPermissionService

因为当前 IPermissionService 的语义非常清晰：

  • 它是“权限布尔检查器”

如果直接把 scope 决策一起塞进去：

  • 现有 controller 依赖会变混乱
  • service 层也很难渐进迁移

更合理的做法是：

  • PermissionServiceImpl 继续做功能权限布尔校验
  • PermissionScopeServiceImpl 负责完整决策

6. Repository 骨架

6.1 permissionGrantRepository.py

职责：

  • 加载用户所有 role / permission grant 明细

建议接口：

class PermissionGrantRepository:
    async def ListUserPermissionGrants(self, UserId: int, PermissionKey: str) -> list[PermissionGrant]:
        ...

    async def GetUserBaseIdentity(self, UserId: int) -> dict[str, Any]:
        ...

推荐 SQL 思路：

SELECT
    r.id AS role_id,
    r.role_key,
    p.permission_key,
    rp.grant_type,
    r.data_scope AS role_scope,
    rp.data_scope AS permission_scope,
    COALESCE(r.priority, 0) AS priority,
    COALESCE(r.is_system_role, FALSE) AS is_system_role,
    rp.condition_filter
FROM user_role ur
JOIN roles r ON r.id = ur.role_id
JOIN role_permissions rp ON rp.role_id = r.id
JOIN permissions p ON p.id = rp.permission_id
WHERE ur.user_id = :user_id

然后在 Python 层执行：

  • 精确匹配
  • wildcard 匹配
  • DENY/GRANT 分类

6.2 resourceScopeRepository.py

职责：

  • 为详情、下载、删除、导出类接口提供主资源回溯

建议接口：

class ResourceScopeRepository:
    async def GetDocumentScopeSnapshot(self, DocumentId: int) -> dict[str, Any] | None: ...
    async def GetGovdocRunScopeSnapshot(self, RunId: int) -> dict[str, Any] | None: ...
    async def GetRagDatasetScopeSnapshot(self, DatasetId: int) -> dict[str, Any] | None: ...
    async def GetCrossReviewProposalScopeSnapshot(self, ProposalId: int) -> dict[str, Any] | None: ...

这些方法不是做最终权限判断，而是提供：

  • 主资源地区
  • 主资源创建者
  • 关联任务/数据集/文档 ID

供 policy 判断使用。

7. Resolver 骨架

7.1 dataScopeResolver.py

职责：

  • 对 PermissionGrant 做范围合并

建议骨架：

class DataScopeResolver:
    _SCOPE_RANK = {
        "SELF": 1,
        "DEPT": 2,
        "ALL": 3,
    }

    def resolve(self, grants: list[PermissionGrant]) -> tuple[bool, str | None, list[str]]:
        exact_denies = [item for item in grants if item.grant_type == "DENY"]
        if exact_denies:
            return False, None, []

        effective_scope = None
        matched_roles: list[str] = []

        for item in grants:
            if item.grant_type != "GRANT":
                continue
            scope = item.permission_scope or item.role_scope or "SELF"
            if effective_scope is None or self._SCOPE_RANK.get(scope, 0) > self._SCOPE_RANK.get(effective_scope, 0):
                effective_scope = scope
            matched_roles.append(item.role_key)

        return True, effective_scope, matched_roles

说明：

  • 第一版先解决 ALL/DEPT/SELF
  • PUBLIC_MIXED、RELATION 不从库里直接出，而由 ModulePolicy 生成

7.2 permissionDecisionService.py

职责：

  • 串联 grant repository、scope resolver、module policy

建议骨架：

class PermissionDecisionService:
    def __init__(
        self,
        grantRepository: PermissionGrantRepository,
        scopeResolver: DataScopeResolver,
        policyRegistry: ModulePolicyRegistry,
    ) -> None:
        self._grantRepository = grantRepository
        self._scopeResolver = scopeResolver
        self._policyRegistry = policyRegistry

    async def Decide(self, Context: ScopeContext) -> PermissionDecision:
        grants = await self._grantRepository.ListUserPermissionGrants(Context.user_id, Context.permission_key)
        allowed, effective_scope, matched_roles = self._scopeResolver.resolve(grants)
        if not allowed:
            return PermissionDecision(
                allowed=False,
                deny_reason="permission_denied",
                effective_scope=None,
                scope_mode="NONE",
            )

        base_decision = PermissionDecision(
            allowed=True,
            deny_reason=None,
            effective_scope=effective_scope,
            scope_mode=effective_scope or "SELF",
            matched_roles=matched_roles,
            matched_permissions=[Context.permission_key],
        )

        policy = self._policyRegistry.Get(Context.module)
        return await policy.Decorate(Context, base_decision)

8. QueryScopeBuilder 骨架

8.1 核心职责

将 PermissionDecision 转成可直接拼接到 SQL 的子句。

8.2 建议骨架

class QueryScopeBuilder:
    def BuildByMapping(
        self,
        Decision: PermissionDecision,
        Mapping: ScopeFieldMapping,
        ScopeUserId: int,
        ScopeArea: str | None,
        RequestedArea: str | None = None,
        RequestedUserId: int | None = None,
    ) -> ScopeClause:
        if not Decision.allowed:
            return ScopeClause(sql="1 = 0", params={}, scope_mode="NONE", description="permission denied")

        if Decision.scope_mode == "ALL":
            if RequestedArea and Mapping.area_field:
                return ScopeClause(
                    sql=f"COALESCE({Mapping.area_field}, '') = :requested_area",
                    params={"requested_area": RequestedArea},
                    scope_mode="ALL",
                    description="global scope with requested area",
                )
            return ScopeClause(sql="1 = 1", params={}, scope_mode="ALL", description="global scope")

        if Decision.scope_mode == "DEPT":
            if not ScopeArea:
                return ScopeClause(sql="1 = 0", params={}, scope_mode="DEPT", description="missing user area")
            if RequestedArea and RequestedArea != ScopeArea:
                return ScopeClause(sql="1 = 0", params={}, scope_mode="DEPT", description="requested area out of scope")
            return ScopeClause(
                sql=f"COALESCE({Mapping.area_field}, '') = :scope_area",
                params={"scope_area": ScopeArea},
                scope_mode="DEPT",
                description="same area scope",
            )

        if Decision.scope_mode == "SELF":
            user_field = Mapping.creator_field or Mapping.owner_field or Mapping.user_field
            if not user_field:
                return ScopeClause(sql="1 = 0", params={}, scope_mode="SELF", description="missing self mapping")
            if RequestedUserId is not None and RequestedUserId != ScopeUserId:
                return ScopeClause(sql="1 = 0", params={}, scope_mode="SELF", description="requested user out of scope")
            return ScopeClause(
                sql=f"{user_field} = :scope_user_id",
                params={"scope_user_id": ScopeUserId},
                scope_mode="SELF",
                description="self scope",
            )

        return ScopeClause(sql="1 = 0", params={}, scope_mode="CUSTOM", description="unsupported generic scope")

9. Policy 骨架

9.1 基础接口

建议文件：policies/base.py

from abc import ABC, abstractmethod

from fastapi_modules.fastapi_leaudit.services.permission_scope.models import PermissionDecision, ScopeClause, ScopeContext


class BaseModulePolicy(ABC):
    @abstractmethod
    async def Decorate(self, Context: ScopeContext, Decision: PermissionDecision) -> PermissionDecision:
        ...

    @abstractmethod
    async def BuildClause(self, Context: ScopeContext, Decision: PermissionDecision) -> ScopeClause | None:
        ...

9.2 DocumentPolicy

建议职责：

  1. 保持 ALL/DEPT/SELF
  2. 固定文档模块字段映射为：
    • d.region
    • f.created_by
  3. 对详情、状态、附件、确认等接口复用同一资源规则

建议骨架：

class DocumentPolicy(BaseModulePolicy):
    def __init__(self, Builder: QueryScopeBuilder) -> None:
        self._builder = Builder

    async def Decorate(self, Context: ScopeContext, Decision: PermissionDecision) -> PermissionDecision:
        return Decision

    async def BuildClause(self, Context: ScopeContext, Decision: PermissionDecision) -> ScopeClause:
        return self._builder.BuildByMapping(
            Decision=Decision,
            Mapping=ScopeFieldMapping(area_field="d.region", creator_field="f.created_by"),
            ScopeUserId=Context.user_id,
            ScopeArea=Context.user_area,
            RequestedArea=Context.request_area,
            RequestedUserId=Context.target_user_id,
        )

9.3 RagPolicy

建议职责：

  1. 读接口转为 PUBLIC_MIXED
  2. 管理接口仍回到 ALL/DEPT/SELF

建议骨架：

class RagPolicy(BaseModulePolicy):
    def __init__(self, Builder: QueryScopeBuilder) -> None:
        self._builder = Builder

    async def Decorate(self, Context: ScopeContext, Decision: PermissionDecision) -> PermissionDecision:
        if Context.permission_key in {"rag:app:read", "rag:chat:use", "rag:dataset:read"}:
            Decision.scope_mode = "PUBLIC_MIXED"
            Decision.allow_public = True
            Decision.allowed_areas = [area for area in [Context.user_area, "省级", ""] if area is not None]
        return Decision

    async def BuildClause(self, Context: ScopeContext, Decision: PermissionDecision) -> ScopeClause:
        if Decision.scope_mode == "PUBLIC_MIXED":
            return ScopeClause(
                sql="(COALESCE(dataset.area, '') IN :visible_areas OR dataset.is_public = TRUE)",
                params={"visible_areas": Decision.allowed_areas},
                scope_mode="PUBLIC_MIXED",
                description="rag public mixed scope",
            )
        return self._builder.BuildByMapping(
            Decision=Decision,
            Mapping=ScopeFieldMapping(area_field="dataset.area", creator_field="dataset.created_by", public_field="dataset.is_public"),
            ScopeUserId=Context.user_id,
            ScopeArea=Context.user_area,
            RequestedArea=Context.request_area,
        )

9.4 CrossReviewPolicy

建议职责：

  • 把普通 SELF/DEPT 转成关系型 RELATION

示意：

class CrossReviewPolicy(BaseModulePolicy):
    async def Decorate(self, Context: ScopeContext, Decision: PermissionDecision) -> PermissionDecision:
        if Decision.allowed:
            Decision.scope_mode = "RELATION"
        return Decision

    async def BuildClause(self, Context: ScopeContext, Decision: PermissionDecision) -> ScopeClause:
        return ScopeClause(
            sql=(
                "EXISTS ("
                "SELECT 1 FROM leaudit_cross_review_task_member tm "
                "WHERE tm.task_id = task.id "
                "AND tm.user_id = :scope_user_id "
                "AND tm.deleted_at IS NULL"
                ")"
            ),
            params={"scope_user_id": Context.user_id},
            scope_mode="RELATION",
            description="cross review relation scope",
        )

10. Facade 骨架

建议文件：facade/scopeAwarePermissionFacade.py

职责：

给各个业务 service 一个最直接的调用入口。

class ScopeAwarePermissionFacade:
    def __init__(
        self,
        DecisionService: PermissionDecisionService,
        PolicyRegistry: ModulePolicyRegistry,
    ) -> None:
        self._decisionService = DecisionService
        self._policyRegistry = PolicyRegistry

    async def Require(self, Context: ScopeContext) -> tuple[PermissionDecision, ScopeClause | None]:
        decision = await self._decisionService.Decide(Context)
        if not decision.allowed:
            raise LeauditException(StatusCodeEnum.HTTP_403_FORBIDDEN, "当前用户没有访问权限")
        policy = self._policyRegistry.Get(Context.module)
        clause = await policy.BuildClause(Context, decision)
        return decision, clause

这样业务 service 的接入就会很轻：

decision, scope_clause = await self.ScopeFacade.Require(...)

11. Controller 接入策略

11.1 第一阶段不强制改 controller

当前 controller 大多已经有：

  • verify_access_token
  • PermissionService.CheckPermission()

第一阶段不建议全部改掉。

建议策略：

  1. controller 继续做功能权限预筛
  2. service 内再做统一 scope 决策

原因：

  • 改动更可控
  • 避免一次性动太多 controller

11.2 第二阶段再统一 controller

等统一执行器稳定后，可以进一步做：

  • controller 不再自行 CheckPermission
  • 改为 service 内一站式处理

但这不是第一批落地的必要条件。

12. documentServiceImpl 接入示例

12.1 当前问题

当前文档模块有：

  • _getCurrentUserContext
  • _buildDocumentScopeFilters

它们本质上就是统一执行器的前身，但还停留在角色硬编码层。

12.2 推荐替换方式

原有思路

currentUser = await self._getCurrentUserContext(CurrentUserId)
filters = self._buildDocumentScopeFilters(...)

改造后思路

context = ScopeContext(
    user_id=CurrentUserId,
    permission_key="documents:list:read",
    module="documents",
    action="read",
    user_area=payload_area,
    request_area=Region,
    target_user_id=UserId,
)
decision, scope_clause = await self.ScopeFacade.Require(context)

然后在 SQL 中：

filters = [
    "d.deleted_at IS NULL",
    "f.deleted_at IS NULL",
    scope_clause.sql,
]
params = {
    **scope_clause.params,
}

12.3 示例骨架

async def ListDocuments(...):
    identity = await self.PermissionGrantRepository.GetUserBaseIdentity(CurrentUserId)
    context = ScopeContext(
        user_id=CurrentUserId,
        permission_key="documents:list:read",
        module="documents",
        action="read",
        user_area=identity.get("area"),
        request_area=Region,
        target_user_id=UserId,
    )
    _, scope_clause = await self.ScopeFacade.Require(context)

    where_clauses = [
        "d.deleted_at IS NULL",
        "f.deleted_at IS NULL",
        scope_clause.sql,
    ]
    params: dict[str, Any] = {
        **scope_clause.params,
    }

    if Keyword:
        where_clauses.append("d.document_name ILIKE :keyword")
        params["keyword"] = f"%{Keyword}%"

    sql = f"""
        SELECT ...
        FROM leaudit_documents d
        JOIN leaudit_document_files f ON f.document_id = d.id
        WHERE {' AND '.join(where_clauses)}
    """

12.4 详情接口示例

async def GetDocument(self, CurrentUserId: int, Id: int) -> DocumentDetailVO:
    identity = await self.PermissionGrantRepository.GetUserBaseIdentity(CurrentUserId)
    context = ScopeContext(
        user_id=CurrentUserId,
        permission_key="documents:detail:read",
        module="documents",
        action="read",
        user_area=identity.get("area"),
        resource_id=Id,
    )
    _, scope_clause = await self.ScopeFacade.Require(context)

    row = (
        await session.execute(
            text(
                f"""
                SELECT ...
                FROM leaudit_documents d
                JOIN leaudit_document_files f ON f.document_id = d.id
                WHERE d.id = :document_id
                  AND {scope_clause.sql}
                """
            ),
            {"document_id": Id, **scope_clause.params},
        )
    ).mappings().first()

重点：

  • 把 scope 直接写进详情查询
  • 不再“先查后判”

13. govdocServiceImpl 接入示例

13.1 当前问题

公文模块里最大风险不是列表，而是：

  • runId
  • report/docx
  • original

这些派生接口。

13.2 推荐接法

先统一一个内部方法：

async def _getScopedGovdocDocument(
    self,
    CurrentUserId: int,
    PermissionKey: str,
    DocumentId: int,
) -> dict[str, Any]:
    ...

所有：

  • GetDocumentDetail
  • CreateRun
  • DownloadOriginal

都先用它。

对 runId 类接口

新增一个仓储回溯：

snapshot = await self.ResourceScopeRepository.GetGovdocRunScopeSnapshot(runId)
document_id = snapshot["document_id"]
document = await self._getScopedGovdocDocument(CurrentUserId, "govdoc:run:read", document_id)

这样所有 run 结果、报告下载都回到 document scope。

14. ragDatasetServiceImpl 接入示例

14.1 当前问题

当前签名大量透传：

  • UserArea
  • UserRole

这就是旧架构的直接暴露。

14.2 第一阶段兼容策略

第一阶段不需要立刻改所有接口签名。

可以先保留：

async def GetAdminDatasets(self, CurrentUserId, UserArea, UserRole, Area, OnlyEnabled, Page, PageSize)

但内部不再使用 UserRole。

示例

async def GetAdminDatasets(...):
    context = ScopeContext(
        user_id=CurrentUserId,
        permission_key="rag:dataset:manage",
        module="rag",
        action="read",
        user_area=UserArea,
        request_area=Area,
    )
    _, scope_clause = await self.ScopeFacade.Require(context)

然后把旧逻辑：

if UserRole not in ("provincial_admin", "admin", "super_admin"):
    raise ...

全部删掉。

14.3 第二阶段签名优化

稳定后再把 service 接口签名收敛为：

async def GetAdminDatasets(self, CurrentUserId: int, Area: str | None, OnlyEnabled: bool | None, Page: int, PageSize: int)

即：

  • UserArea
  • UserRole

不再从 controller 显式透传。

而是在 service 内统一从 identity/repository 获取。

14.4 数据集子资源示例

async def GetDatasetDocumentDetail(...):
    identity = await self.PermissionGrantRepository.GetUserBaseIdentity(CurrentUserId)
    context = ScopeContext(
        user_id=CurrentUserId,
        permission_key="rag:dataset:read",
        module="rag",
        action="read",
        user_area=identity.get("area"),
        resource_id=DatasetId,
    )
    _, dataset_scope_clause = await self.ScopeFacade.Require(context)

    sql = f"""
        SELECT ...
        FROM rag_datasets dataset
        JOIN rag_documents doc ON doc.dataset_id = dataset.id
        WHERE dataset.id = :dataset_id
          AND doc.id = :document_id
          AND {dataset_scope_clause.sql}
    """

重点：

  • 文档详情跟随 dataset scope
  • 不再自己判断角色

15. rbacAdminServiceImpl 接入示例

15.1 当前问题

当前 _assertManagePermission 和 _assertPermission 双轨并存。

还存在：

bool_or(r.role_key IN ('super_admin', 'provincial_admin', 'admin')) AS can_manage

15.2 推荐做法

角色管理对象本身和用户管理对象分开处理。

角色元数据接口

例如：

  • ListRoles
  • CreateRole
  • SaveRolePermissions

主要按功能权限控制，不做 area scope。

用户相关接口

例如：

  • ListUsers
  • AssignUserRoles
  • GetUserRoles

要做用户 area scope。

15.3 ListUsers 示例

async def ListUsers(...):
    identity = await self.PermissionGrantRepository.GetUserBaseIdentity(CurrentUserId)
    context = ScopeContext(
        user_id=CurrentUserId,
        permission_key="rbac:users:read",
        module="rbac_admin",
        action="read",
        user_area=identity.get("area"),
        request_area=Area,
    )
    _, scope_clause = await self.ScopeFacade.Require(context)

    sql = f"""
        SELECT ...
        FROM sso_users u
        WHERE u.deleted_at IS NULL
          AND {scope_clause.sql}
    """

RbacAdminPolicy 内固定映射：

  • area_field = u.area

15.4 AssignUserRoles 示例

建议先做用户合法性校验：

target_context = ScopeContext(
    user_id=CurrentUserId,
    permission_key="rbac:user_roles:write",
    module="rbac_admin",
    action="write",
    user_area=identity.get("area"),
    resource_id=UserId,
)
_, scope_clause = await self.ScopeFacade.Require(target_context)

然后先查：

SELECT u.id
FROM sso_users u
WHERE u.id = :target_user_id
  AND {scope_clause.sql}

只有通过后才执行角色绑定。

16. 接口签名迁移建议

16.1 第一阶段

原则：

  • 不强行修改所有抽象接口签名
  • 优先兼容现有 controller

例如 RAG service 可以暂时保留：

  • UserArea
  • UserRole

但内部只真正使用：

  • CurrentUserId
  • UserArea

并逐步废弃 UserRole

16.2 第二阶段

当统一执行器稳定后，建议收敛为：

  1. controller 只传 CurrentUserId
  2. service 自己取基础身份
  3. ScopeContext 统一构造

这一步是“收口优化”，不是第一批落地阻塞项。

17. 日志与异常建议

17.1 统一异常

建议新增：

class PermissionScopeDeniedException(LeauditException):
    ...

用于区分：

  • 功能权限拒绝
  • 数据范围拒绝
  • 资源不存在

17.2 日志建议

每次 scope 拒绝建议记录：

logger.warning(
    "scope denied",
    extra={
        "user_id": Context.user_id,
        "permission_key": Context.permission_key,
        "module": Context.module,
        "resource_id": Context.resource_id,
        "request_area": Context.request_area,
        "scope_mode": decision.scope_mode,
        "deny_reason": decision.deny_reason,
    },
)

这对后续排查误拒绝和越权都很重要。

18. 单元测试骨架建议

建议至少补以下测试：

DataScopeResolver

  • GRANT SELF + GRANT DEPT -> DEPT
  • GRANT ALL + DENY exact -> denied
  • permission_scope 覆盖 role_scope

QueryScopeBuilder

  • ALL
  • DEPT
  • SELF
  • SELF + requested_user_id != scope_user_id

RagPolicy

  • rag:dataset:read -> PUBLIC_MIXED
  • rag:dataset:update -> 保持 ALL/DEPT/SELF

CrossReviewPolicy

  • allowed -> RELATION

19. 推荐实施顺序

代码落地建议按下面顺序推进：

  1. 建 models / resolver / builder / policy base
  2. 建 DocumentPolicy
  3. 用文档模块先跑通一条完整链路
  4. 建 RagPolicy，清理 RAG 角色白名单
  5. 建 GovdocPolicy，收口 run/report/download
  6. 建 UsageStatsPolicy
  7. 建 RbacAdminPolicy
  8. 最后收敛 service 签名

这样做的好处是：

  • 先拿最典型模块验证设计
  • 再扩到复杂特例
  • 避免一开始就改满全仓库

20. 最终判断标准

这份代码骨架真正落地完成，不是看文件建没建，而是看下面几件事是否成立：

  1. 文档模块不再依赖 _getCurrentUserContext + role_key IN (...)
  2. RAG 模块不再依赖 UserRole 白名单
  3. 公文下载/报告接口已通过主资源回溯做 scope 判断
  4. RBAC 用户管理域不再通过角色名派生 can_manage
  5. 新增一个接口时，研发可以直接按 ScopeContext -> Require -> ScopeClause 方式接入

如果还做不到第 5 条，说明统一执行器还没有真正成为平台能力。

Reference in New Issue View Git Blame Copy Permalink