TanWenyan/leaudit-platform-backend
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(auth): enforce document and govdoc route grants

Browse Source
This commit is contained in:
wren
2026-05-25 15:37:53 +08:00
parent 75c077da77
commit 4ac53ded5a
8 changed files with 424 additions and 19 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_document_type_permissions.py
+169
View File
@@ -0,0 +1,169 @@
"""文档类型权限控制测试。"""
import pytest
from starlette.responses import JSONResponse
from fastapi_modules.fastapi_leaudit.controllers.documentController import DocumentController
from fastapi_modules.fastapi_leaudit.domian.vo.documentVo import (
DocumentTypeCreateDTO,
DocumentTypeItemVO,
DocumentTypeRootCreateDTO,
DocumentTypeRootItemVO,
DocumentTypeRootUpdateDTO,
DocumentTypeUpdateDTO,
)
class _DenyPermissionService:
"""拒绝所有权限的测试权限服务。"""
async def CheckPermission(self, UserId: int, PermissionKey: str) -> bool:
return False
class _AllowOnlyPermissionService:
"""只允许指定权限的测试权限服务。"""
def __init__(self, allowed: set[str]) -> None:
self.allowed = allowed
async def CheckPermission(self, UserId: int, PermissionKey: str) -> bool:
return PermissionKey in self.allowed
class _FakeDocumentService:
"""记录调用的测试文档服务。"""
def __init__(self) -> None:
self.calls: list[str] = []
async def ListDocumentTypes(self, **kwargs):
self.calls.append("ListDocumentTypes")
return [
DocumentTypeItemVO(
id=1,
name="合同",
code="contract",
description=None,
entryModuleId=None,
isEnabled=True,
ruleSetIds=[],
)
]
async def GetDocumentType(self, **kwargs):
self.calls.append("GetDocumentType")
return DocumentTypeItemVO(
id=1,
name="合同",
code="contract",
description=None,
entryModuleId=None,
isEnabled=True,
ruleSetIds=[],
)
async def CreateDocumentType(self, **kwargs):
self.calls.append("CreateDocumentType")
return await self.GetDocumentType()
async def UpdateDocumentType(self, **kwargs):
self.calls.append("UpdateDocumentType")
return await self.GetDocumentType()
async def DeleteDocumentType(self, **kwargs):
self.calls.append("DeleteDocumentType")
async def ListDocumentTypeRoots(self, **kwargs):
self.calls.append("ListDocumentTypeRoots")
return [
DocumentTypeRootItemVO(
id=11,
name="合同",
code="root.contract",
description=None,
entryModuleId=None,
entryModuleName=None,
isEnabled=True,
childGroupCount=0,
ruleSetCount=0,
ruleSetIds=[],
)
]
async def GetDocumentTypeRoot(self, **kwargs):
self.calls.append("GetDocumentTypeRoot")
return DocumentTypeRootItemVO(
id=11,
name="合同",
code="root.contract",
description=None,
entryModuleId=None,
entryModuleName=None,
isEnabled=True,
childGroupCount=0,
ruleSetCount=0,
ruleSetIds=[],
)
async def CreateDocumentTypeRoot(self, **kwargs):
self.calls.append("CreateDocumentTypeRoot")
return await self.GetDocumentTypeRoot()
async def UpdateDocumentTypeRoot(self, **kwargs):
self.calls.append("UpdateDocumentTypeRoot")
return await self.GetDocumentTypeRoot()
def _find_endpoint(controller: DocumentController, path: str, method: str):
"""根据路径和方法查找路由 endpoint。"""
full_path = f"{controller.router.prefix}{path}"
for route in controller.router.routes:
if getattr(route, "path", "") == full_path and method in getattr(route, "methods", set()):
return route.endpoint
raise AssertionError(f"未找到路由 {method} {full_path}")
@pytest.mark.asyncio
@pytest.mark.parametrize(
("path", "method", "kwargs", "expected_call"),
[
("/document-types", "GET", {"ids": None, "entry_module_id": None}, "ListDocumentTypes"),
("/document-types/{TypeId}", "GET", {"TypeId": 1}, "GetDocumentType"),
("/document-types", "POST", {"Body": DocumentTypeCreateDTO(code="contract", name="合同")}, "CreateDocumentType"),
("/document-types/{TypeId}", "PUT", {"TypeId": 1, "Body": DocumentTypeUpdateDTO(name="合同")}, "UpdateDocumentType"),
("/document-types/{TypeId}", "DELETE", {"TypeId": 1}, "DeleteDocumentType"),
("/v3/document-type-roots", "GET", {"entry_module_id": None}, "ListDocumentTypeRoots"),
("/v3/document-type-roots/{RootId}", "GET", {"RootId": 11}, "GetDocumentTypeRoot"),
("/v3/document-type-roots", "POST", {"Body": DocumentTypeRootCreateDTO(code="root.contract", name="合同")}, "CreateDocumentTypeRoot"),
("/v3/document-type-roots/{RootId}", "PUT", {"RootId": 11, "Body": DocumentTypeRootUpdateDTO(name="合同")}, "UpdateDocumentTypeRoot"),
],
)
async def test_document_type_endpoints_require_permission(path, method, kwargs, expected_call):
"""文档类型和业务大类接口无权限时返回 403，且不调用业务服务。"""
controller = DocumentController()
service = _FakeDocumentService()
controller.DocumentService = service
controller.PermissionService = _DenyPermissionService()
endpoint = _find_endpoint(controller, path, method)
response = await endpoint(**kwargs, payload={"user_id": 7})
assert isinstance(response, JSONResponse)
assert response.status_code == 403
assert expected_call not in service.calls
@pytest.mark.asyncio
async def test_document_type_root_list_calls_service_when_permission_granted():
"""业务大类列表有查看权限时正常调用业务服务。"""
controller = DocumentController()
service = _FakeDocumentService()
controller.DocumentService = service
controller.PermissionService = _AllowOnlyPermissionService({"doc_type:list:read"})
endpoint = _find_endpoint(controller, "/v3/document-type-roots", "GET")
response = await endpoint(entry_module_id=None, payload={"user_id": 7})
assert response.data[0].id == 11
assert service.calls == ["ListDocumentTypeRoots"]
tests/test_govdoc_permissions.py
+88
View File
@@ -145,3 +145,91 @@ def test_govdoc_root_route_marks_frontend_route_set_ready():
]
assert service._isFrontendRouteSetReady(routes) is True
def test_govdoc_parent_route_does_not_expose_ungranted_child_routes():
"""只有内部公文父路由和接口权限时，不应补出未勾选的列表/上传子路由。"""
service = RbacServiceImpl()
routes = [
RbacRouteVO(
id=1,
route_path="/govdoc",
route_name="govdoc",
component="govdoc",
parent_id=None,
route_title="内部公文处理",
children=[
RbacRouteVO(
id=2,
route_path="/govdoc/audits",
route_name="govdoc-audits",
component="govdoc.audits",
parent_id=1,
route_title="公文列表",
),
RbacRouteVO(
id=3,
route_path="/govdoc/upload",
route_name="govdoc-upload",
component="govdoc.upload",
parent_id=1,
route_title="公文上传",
),
],
)
]
filtered = service._filterRoutesByRouteAndPermissionScope(
routes,
{"/govdoc"},
{"govdoc:document:read", "govdoc:document:create"},
)
paths = service._collectRoutePaths(filtered)
assert "/govdoc" in paths
assert "/govdoc/audits" not in paths
assert "/govdoc/upload" not in paths
def test_legacy_govdoc_audit_route_does_not_grant_current_govdoc_child_route():
"""旧 /govdoc-audit 残留授权不应继续放行当前 /govdoc 子路由。"""
service = RbacServiceImpl()
routes = [
RbacRouteVO(
id=1,
route_path="/govdoc",
route_name="govdoc",
component="govdoc",
parent_id=None,
route_title="内部公文处理",
),
RbacRouteVO(
id=2,
route_path="/govdoc-audit/audits",
route_name="legacy-govdoc-audits",
component="govdoc-audit.audits",
parent_id=None,
route_title="旧公文列表",
),
RbacRouteVO(
id=3,
route_path="/govdoc-audit/upload",
route_name="legacy-govdoc-upload",
component="govdoc-audit.upload",
parent_id=None,
route_title="旧公文上传",
),
]
filtered = service._filterRoutesByRouteAndPermissionScope(
routes,
service._collectCurrentFrontendRoutePaths(routes),
{"govdoc:document:read", "govdoc:document:create"},
)
paths = service._collectRoutePaths(filtered)
assert "/govdoc" in paths
assert "/govdoc-audit/audits" not in paths
assert "/govdoc-audit/upload" not in paths
assert "/govdoc/audits" not in paths
assert "/govdoc/upload" not in paths
tests/test_rule_write_scope.py
+7 -5
View File
@@ -183,7 +183,7 @@ def test_rbac_manageable_permissions_include_rule_version_lifecycle():
assert "rules:binding_delete:delete" in permission_keys
def test_rbac_rule_group_permissions_are_folded_into_rules_menu():
def test_rbac_rule_groups_route_is_exposed_under_settings():
route_paths = {item["route_path"] for item in RbacAdminServiceImpl._MANAGEABLE_ROUTE_BLUEPRINTS}
group_permission_paths = {
item["route_path"]
@@ -191,17 +191,19 @@ def test_rbac_rule_group_permissions_are_folded_into_rules_menu():
if item["permission_key"].startswith("evaluation_group:")
}
assert "/rule-groups" not in route_paths
assert "/rule-groups" in route_paths
assert group_permission_paths == {"/rules"}
def test_user_route_compat_menu_does_not_expose_rule_groups():
def test_user_route_compat_menu_exposes_rule_groups_under_settings():
service = RbacServiceImpl()
routes = service._buildCompatibilityRoutes(["admin"], {"evaluation_group:list:read", "rules:list:read"})
paths = service._collectRoutePaths(routes)
rules_route = next(route for route in routes if route.route_path == "/rules")
settings_route = next(route for route in routes if route.route_path == "/settings")
rule_groups_route = next(route for route in (settings_route.children or []) if route.route_path == "/rule-groups")
assert "/rule-groups" not in paths
assert "/rule-groups" in paths
assert rule_groups_route.parent_id == settings_route.id
assert "evaluation_group:list:read" in rules_route.permissions