"""需求16 矩阵判定融合测试(authorization_service)。 覆盖: - OWN/OTHER 部门两分类矩阵判定(OTHER 涵盖所属部门子树外全部部门) - 矩阵与单文档 ACL 的 OR 叠加 - 密级前置与状态前置不可被矩阵绕过 - 无角色→默认角色→无默认则贡献为空 - 显式角色停用后矩阵贡献为空且不回退默认 - plan_allowed_actions 的矩阵操作位叠加(子方案过滤授权/挂载位) - 受限管理员(非内置 ADMIN)文档权限同 USER 按矩阵判定,超管保持全权 """ from __future__ import annotations from datetime import datetime, timedelta, timezone import pytest from dms.common.enums import AllowedAction, RoleCode, SecurityLevel from dms.common.errors import ( DocumentActionForbiddenError, SecurityLevelForbiddenError, ) from dms.extensions import db from dms.models import ( Category, Document, Organization, Permission, Role, RolePermission, User, ) from dms.security.auth_context import AuthContext from dms.services.authorization_service import ( evaluate_plan_access, plan_allowed_actions, require_plan_action, ) def _make_document(name: str, *, org: Organization, security: str = "SECRET", status: str = "PUBLISHED", document_type: str = "MAIN", parent: Document | None = None, root: Document | None = None, attachment_type: str | None = None) -> Document: # shape v3(迁移0026):MAIN/SUB_PLAN 必须挂分类且无附件类型, # ATTACHMENT 必须有附件类型且不挂分类。懒建专用分类补齐约束字段。 category = db.session.scalar( db.select(Category).where(Category.category_code == "MATRIX_TEST") ) if category is None: category = Category( category_code="MATRIX_TEST", category_name="矩阵测试分类", category_type="OTHER", category_path="/矩阵测试分类", sort_no=99, document_count=0, status="ENABLED", ) db.session.add(category) db.session.flush() item = Document( document_name=name, summary=f"{name}摘要", document_type=document_type, document_status=status, security_level=security, attachment_type=attachment_type, organization_id=org.id, organization_name=org.org_name, category_id=( category.id if document_type != "ATTACHMENT" else None ), category_name=( category.category_name if document_type != "ATTACHMENT" else None ), category_path=( category.category_path if document_type != "ATTACHMENT" else None ), parent_document_id=parent.id if parent else None, root_document_id=root.id if root else None, tags=["矩阵测试"], original_file_name=f"{name}.docx", file_relative_path=f"original/MATRIX_{name}.docx", file_extension="docx", mime_type="application/vnd.openxmlformats-officedocument.wordprocessingml.document", file_size=128, file_hash="b" * 64, search_text=f"{name} 矩阵测试", created_by_name="系统管理员", updated_by_name="系统管理员", ) db.session.add(item) db.session.flush() return item def _make_role(name: str, *, is_default: bool = False, status: str = "ENABLED", actor_id: int | None = None) -> Role: role = Role( role_name=name, sort_no=10, status=status, is_default=is_default, created_by=actor_id, updated_by=actor_id, ) db.session.add(role) db.session.flush() return role def _matrix_row(role_id: int, level: str, scope: str, **flags) -> None: db.session.add( RolePermission( role_id=role_id, security_level=level, dept_scope=scope, **flags, ) ) def _user_context(user: User) -> AuthContext: return AuthContext( user_id=user.id, username=user.username, real_name=user.real_name, organization_id=user.organization_id, organization_name=user.organization_name, role_code=RoleCode(user.role_code), security_level=SecurityLevel(user.security_level), auth_version=user.auth_version, ) @pytest.fixture() def matrix_env(b2_app, role_cleanup): """在 b2_app 基础上构造矩阵判定数据:机构、用户、文档。""" with b2_app.app_context(): ops = db.session.scalar( db.select(Organization).where(Organization.org_code == "ORG_OPS") ) comms = db.session.scalar( db.select(Organization).where(Organization.org_code == "ORG_COMMS") ) user = db.session.scalar( db.select(User).where(User.username == "user") ) assert ops and comms and user db.session.commit() ids = {"ops": ops.id, "comms": comms.id, "user": user.id} return b2_app, ids @pytest.mark.integration def test_own_scope_matrix_grants_view_when_acl_denies(matrix_env): app, ids = matrix_env with app.app_context(): ops = db.session.get(Organization, ids["ops"]) user = db.session.get(User, ids["user"]) role = _make_role("作战部角色", actor_id=user.id) user.role_id = role.id _matrix_row( role.id, "SECRET", "OWN", can_view=True, can_download=True ) # 无借阅 ACL 行:纯矩阵放行 doc = _make_document("矩阵本部门主案", org=ops) db.session.commit() context = _user_context(user) access = evaluate_plan_access(doc, context) assert access.allowed is True assert access.can_download is True assert "VIEW" in access.matrix_actions assert "DOWNLOAD" in access.matrix_actions @pytest.mark.integration def test_other_scope_without_matrix_denies(matrix_env): app, ids = matrix_env with app.app_context(): comms = db.session.get(Organization, ids["comms"]) user = db.session.get(User, ids["user"]) # 文档挂通信部:对作战部用户是 OTHER,且未配置部门 doc = _make_document("矩阵其他部门主案", org=comms) db.session.commit() context = _user_context(user) access = evaluate_plan_access(doc, context) assert access.allowed is False assert access.reason == "ACL" @pytest.mark.integration def test_other_scope_matrix_grants_cross_department_view(matrix_env): app, ids = matrix_env with app.app_context(): comms = db.session.get(Organization, ids["comms"]) user = db.session.get(User, ids["user"]) role = _make_role("跨部门角色", actor_id=user.id) user.role_id = role.id _matrix_row(role.id, "SECRET", "OTHER", can_view=True) doc = _make_document("矩阵其他部门主案", org=comms) db.session.commit() context = _user_context(user) access = evaluate_plan_access(doc, context) assert access.allowed is True assert "VIEW" in access.matrix_actions @pytest.mark.integration def test_security_and_status_precedence_over_matrix(matrix_env): app, ids = matrix_env with app.app_context(): ops = db.session.get(Organization, ids["ops"]) user = db.session.get(User, ids["user"]) # 密级 SECRET role = _make_role("越权角色", actor_id=user.id) _matrix_row(role.id, "TOP_SECRET", "OWN", can_view=True) top_secret_doc = _make_document( "矩阵绝密主案", org=ops, security="TOP_SECRET" ) draft_doc = _make_document( "矩阵草稿主案", org=ops, security="SECRET", status="DRAFT" ) db.session.commit() context = _user_context(user) assert evaluate_plan_access(top_secret_doc, context).reason == "SECURITY" assert evaluate_plan_access(draft_doc, context).reason == "STATUS" @pytest.mark.integration def test_default_role_applies_and_disabled_role_yields_nothing(matrix_env): app, ids = matrix_env with app.app_context(): ops = db.session.get(Organization, ids["ops"]) user = db.session.get(User, ids["user"]) # 未挂角色 default_role = _make_role("默认角色", is_default=True, actor_id=user.id) _matrix_row(default_role.id, "SECRET", "OWN", can_view=True) doc = _make_document("默认角色主案", org=ops) db.session.commit() context = _user_context(user) assert evaluate_plan_access(doc, context).allowed is True # 默认角色停用:无显式角色且无默认 → 矩阵贡献为空 default_role.status = "DISABLED" db.session.commit() denied = evaluate_plan_access(doc, context) assert denied.allowed is False # 显式挂停用角色:不回退默认角色 disabled_role = _make_role( "停用角色", status="DISABLED", actor_id=user.id ) _matrix_row(disabled_role.id, "SECRET", "OWN", can_view=True) user.role_id = disabled_role.id db.session.commit() assert evaluate_plan_access(doc, context).allowed is False @pytest.mark.integration def test_matrix_actions_expand_allowed_actions(matrix_env): app, ids = matrix_env with app.app_context(): ops = db.session.get(Organization, ids["ops"]) user = db.session.get(User, ids["user"]) role = _make_role("完整权限角色", actor_id=user.id) user.role_id = role.id _matrix_row( role.id, "SECRET", "OWN", can_view=True, can_download=True, can_edit=True, can_delete=True, can_config_permission=True, can_bind_attachment=True, can_unbind_attachment=True, ) main_doc = _make_document("矩阵操作位主案", org=ops) sub_doc = _make_document( "矩阵操作位子案", org=ops, document_type="SUB_PLAN", parent=main_doc, root=main_doc, ) db.session.commit() context = _user_context(user) main_actions = plan_allowed_actions( main_doc, context, evaluate_plan_access(main_doc, context) ) assert set(main_actions) >= { "VIEW", "DOWNLOAD", "EDIT", "DELETE", "CONFIG_PERMISSION", } sub_actions = plan_allowed_actions( sub_doc, context, evaluate_plan_access(sub_doc, context) ) # 子方案过滤授权/挂载/解除挂载 assert "CONFIG_PERMISSION" not in sub_actions assert "BIND_ATTACHMENT" not in sub_actions assert "UNBIND_ATTACHMENT" not in sub_actions assert {"VIEW", "DOWNLOAD", "EDIT", "DELETE"} <= set(sub_actions) # ===== 动态管理菜单(allowedModules,契约1.35菜单口径) ======================== # 菜单可见性完全由角色权限决定(纯矩阵/管理位判定,不查文档存在性); # 页签内按钮与接口鉴权仍按 evaluate_plan_access 实时判定。 from dms.services.user_service import allowed_modules_for_user @pytest.mark.integration def test_builtin_admin_keeps_full_modules(matrix_env): app, ids = matrix_env with app.app_context(): admin = db.session.scalar( db.select(User).where(User.username == "admin") ) assert allowed_modules_for_user(admin) == [ "BACKEND_MANAGEMENT", "AUDIT_LOG", "ORGANIZATION_USER_MANAGEMENT", "SYSTEM_MANAGEMENT", ] @pytest.mark.integration def test_view_only_matrix_renders_browser_only(matrix_env): app, ids = matrix_env with app.app_context(): user = db.session.get(User, ids["user"]) role = _make_role("只读角色", is_default=True, actor_id=user.id) # 矩阵只有查看/下载位,无任何管理类位 for scope in ("OWN", "OTHER"): _matrix_row( role.id, "SECRET", scope, can_view=True, can_download=True ) db.session.commit() assert allowed_modules_for_user(user) == ["DOCUMENT_BROWSER"] @pytest.mark.integration def test_manage_matrix_renders_backend_instead_of_browser(matrix_env): app, ids = matrix_env with app.app_context(): user = db.session.get(User, ids["user"]) role = _make_role("编辑角色", is_default=True, actor_id=user.id) _matrix_row( role.id, "SECRET", "OWN", can_view=True, can_edit=True ) db.session.commit() # 管理类列命中:渲染方案管理,不再渲染方案浏览 assert allowed_modules_for_user(user) == ["BACKEND_MANAGEMENT"] @pytest.mark.integration def test_upload_only_matrix_renders_backend(matrix_env): app, ids = matrix_env with app.app_context(): user = db.session.get(User, ids["user"]) role = _make_role("仅上传角色", is_default=True, actor_id=user.id) # 仅勾上传位,无查看/下载/编辑等任何其他位 _matrix_row(role.id, "SECRET", "OWN", can_upload=True) db.session.commit() assert allowed_modules_for_user(user) == ["BACKEND_MANAGEMENT"] @pytest.mark.integration def test_other_scope_manage_bit_without_documents_renders_backend(matrix_env): """纯矩阵判定:OTHER 行勾管理位同样命中,与文档归属/存在无关。""" app, ids = matrix_env with app.app_context(): user = db.session.get(User, ids["user"]) role = _make_role("跨部门编辑角色", is_default=True, actor_id=user.id) # 仅 OTHER × TOP_SECRET 勾编辑:不创建任何文档也应命中(空库不丢菜单) _matrix_row( role.id, "TOP_SECRET", "OTHER", can_view=True, can_edit=True ) db.session.commit() assert allowed_modules_for_user(user) == ["BACKEND_MANAGEMENT"] @pytest.mark.integration def test_manage_bit_without_view_still_renders_backend(matrix_env): """编辑位开启但查看位全关:菜单口径仍按管理类列命中。""" app, ids = matrix_env with app.app_context(): user = db.session.get(User, ids["user"]) role = _make_role("无查看编辑角色", is_default=True, actor_id=user.id) _matrix_row(role.id, "SECRET", "OWN", can_edit=True) db.session.commit() assert allowed_modules_for_user(user) == ["BACKEND_MANAGEMENT"] @pytest.mark.integration def test_empty_matrix_and_bits_falls_back_to_browser(matrix_env): app, ids = matrix_env with app.app_context(): user = db.session.get(User, ids["user"]) # 角色存在但矩阵与管理位全空:兜底渲染方案浏览 _make_role("空角色", is_default=True, actor_id=user.id) db.session.commit() assert allowed_modules_for_user(user) == ["DOCUMENT_BROWSER"] @pytest.mark.integration def test_user_role_manage_bits_get_no_directory_menu(matrix_env): """管理位仅 ADMIN 类别生效:USER 类别勾管理位不出机构用户页签。""" app, ids = matrix_env with app.app_context(): user = db.session.get(User, ids["user"]) role = _make_role("USER类别管理位角色", is_default=True, actor_id=user.id) role.manage_org_edit = True role.manage_user_edit = True role.manage_role_edit = True _matrix_row(role.id, "SECRET", "OWN", can_view=True) db.session.commit() # 管理位被类别限制拦下:仅方案浏览 assert allowed_modules_for_user(user) == ["DOCUMENT_BROWSER"] # ===== 写路径集中鉴权 require_plan_action ==================================== @pytest.mark.integration def test_require_plan_action_allows_matrix_edit(matrix_env): app, ids = matrix_env with app.app_context(): ops = db.session.get(Organization, ids["ops"]) user = db.session.get(User, ids["user"]) role = _make_role("写路径编辑角色", is_default=True, actor_id=user.id) _matrix_row(role.id, "SECRET", "OWN", can_view=True, can_edit=True) doc = _make_document("写路径可编辑主案", org=ops) db.session.commit() require_plan_action(doc, _user_context(user), AllowedAction.EDIT) @pytest.mark.integration def test_require_plan_action_denies_without_bit(matrix_env): app, ids = matrix_env with app.app_context(): ops = db.session.get(Organization, ids["ops"]) user = db.session.get(User, ids["user"]) role = _make_role("写路径只读角色", is_default=True, actor_id=user.id) _matrix_row(role.id, "SECRET", "OWN", can_view=True, can_download=True) doc = _make_document("写路径只读主案", org=ops) db.session.commit() with pytest.raises(DocumentActionForbiddenError): require_plan_action(doc, _user_context(user), AllowedAction.EDIT) with pytest.raises(DocumentActionForbiddenError): require_plan_action(doc, _user_context(user), AllowedAction.DELETE) with pytest.raises(DocumentActionForbiddenError): require_plan_action( doc, _user_context(user), AllowedAction.CONFIG_PERMISSION ) @pytest.mark.integration def test_require_plan_action_security_gate_precedes_bits(matrix_env): app, ids = matrix_env with app.app_context(): ops = db.session.get(Organization, ids["ops"]) user = db.session.get(User, ids["user"]) role = _make_role("写路径高密角色", is_default=True, actor_id=user.id) _matrix_row(role.id, "TOP_SECRET", "OWN", can_view=True, can_edit=True) # 用户密级 SECRET:即使矩阵在 TOP_SECRET 行勾了编辑,密级前置直接拒绝 doc = _make_document("绝密主案", org=ops, security="TOP_SECRET") db.session.commit() with pytest.raises(SecurityLevelForbiddenError): require_plan_action(doc, _user_context(user), AllowedAction.EDIT) @pytest.mark.integration def test_require_plan_action_admin_passes_all(matrix_env): app, ids = matrix_env with app.app_context(): ops = db.session.get(Organization, ids["ops"]) admin = db.session.scalar( db.select(User).where(User.username == "admin") ) doc = _make_document("管理员主案", org=ops, security="SECRET") db.session.commit() for action in ( AllowedAction.EDIT, AllowedAction.DELETE, AllowedAction.CONFIG_PERMISSION, AllowedAction.BIND_ATTACHMENT, AllowedAction.UNBIND_ATTACHMENT, ): require_plan_action(doc, _user_context(admin), action) @pytest.mark.integration def test_require_plan_action_attachment_admin_only(matrix_env): app, ids = matrix_env with app.app_context(): ops = db.session.get(Organization, ids["ops"]) user = db.session.get(User, ids["user"]) admin = db.session.scalar( db.select(User).where(User.username == "admin") ) role = _make_role("附件全权角色", is_default=True, actor_id=user.id) for scope in ("OWN", "OTHER"): _matrix_row( role.id, "PUBLIC", scope, can_view=True, can_download=True, can_edit=True, can_delete=True, ) att = _make_document( "共享附件", org=ops, security="PUBLIC", document_type="ATTACHMENT", attachment_type="TRAINING_MATERIAL", ) db.session.commit() with pytest.raises(DocumentActionForbiddenError): require_plan_action(att, _user_context(user), AllowedAction.EDIT) require_plan_action(att, _user_context(admin), AllowedAction.DELETE) @pytest.mark.integration def test_require_plan_action_allows_edit_without_view(matrix_env): """契约1.36完全拆分:仅勾编辑(未勾查看)时编辑位独立生效。""" app, ids = matrix_env with app.app_context(): ops = db.session.get(Organization, ids["ops"]) user = db.session.get(User, ids["user"]) role = _make_role("无查看编辑角色", is_default=True, actor_id=user.id) _matrix_row(role.id, "SECRET", "OWN", can_edit=True) doc = _make_document("无查看可编辑主案", org=ops) db.session.commit() context = _user_context(user) require_plan_action(doc, context, AllowedAction.EDIT) access = evaluate_plan_access(doc, context) # 查看/下载均未命中:操作位清单只含 EDIT assert plan_allowed_actions(doc, context, access) == ["EDIT"] with pytest.raises(DocumentActionForbiddenError): require_plan_action(doc, context, AllowedAction.DELETE) @pytest.mark.integration def test_download_bit_without_view(matrix_env): """契约1.36完全拆分:仅勾下载(未勾查看)时下载位独立生效。""" app, ids = matrix_env with app.app_context(): ops = db.session.get(Organization, ids["ops"]) user = db.session.get(User, ids["user"]) role = _make_role("无查看下载角色", is_default=True, actor_id=user.id) _matrix_row(role.id, "SECRET", "OWN", can_download=True) doc = _make_document("无查看可下载主案", org=ops) db.session.commit() context = _user_context(user) access = evaluate_plan_access(doc, context) assert access.allowed is False assert access.can_download is True assert plan_allowed_actions(doc, context, access) == ["DOWNLOAD"] @pytest.mark.integration def test_acl_download_without_view(matrix_env): """契约1.36完全拆分:借阅ACL只勾下载(未勾查看)同样独立生效。""" app, ids = matrix_env with app.app_context(): ops = db.session.get(Organization, ids["ops"]) user = db.session.get(User, ids["user"]) _make_role("ACL无矩阵角色", is_default=True, actor_id=user.id) doc = _make_document("ACL下载主案", org=ops) db.session.add( Permission( document_id=doc.id, subject_id=user.id, subject_name=user.real_name, can_view=False, can_download=True, expires_at=( datetime.now(timezone.utc).replace(tzinfo=None) + timedelta(days=1) ), created_by=user.id, updated_by=user.id, ) ) db.session.commit() context = _user_context(user) access = evaluate_plan_access(doc, context) assert access.allowed is False assert access.can_download is True assert plan_allowed_actions(doc, context, access) == ["DOWNLOAD"] # ===== 受限管理员(非内置 ADMIN)文档权限按角色矩阵判定 ======================== from dms.security.passwords import hash_password def _make_limited_admin(org: Organization, *, username: str = "lim_admin", role_id: int | None = None) -> User: """非内置管理员:role_code=ADMIN 且 username != admin,可关联角色。""" admin = User( username=username, password_hash=hash_password("Initial@2026"), real_name="受限管理员", organization_id=org.id, organization_name=org.org_name, role_code="ADMIN", security_level="SECRET", status="ENABLED", role_id=role_id, ) db.session.add(admin) db.session.flush() return admin @pytest.mark.integration def test_limited_admin_actions_follow_matrix(matrix_env): app, ids = matrix_env with app.app_context(): ops = db.session.get(Organization, ids["ops"]) role = _make_role("受限管理员角色") lim_admin = _make_limited_admin(ops, role_id=role.id) _matrix_row( role.id, "SECRET", "OWN", can_view=True, can_download=True, can_edit=True, ) doc = _make_document("受限管理员矩阵主案", org=ops) db.session.commit() context = _user_context(lim_admin) access = evaluate_plan_access(doc, context) assert access.allowed is True # 矩阵授予什么就有什么:未授予的删除/授权/挂载位不再因 ADMIN 身份放行 assert set(plan_allowed_actions(doc, context, access)) == { "VIEW", "DOWNLOAD", "EDIT", } require_plan_action(doc, context, AllowedAction.EDIT) with pytest.raises(DocumentActionForbiddenError): require_plan_action(doc, context, AllowedAction.DELETE) # 管理菜单按纯矩阵推导(契约1.35):勾了 EDIT 即渲染方案管理; # 管理位独立于文档矩阵(契约1.23):该角色未授 manage_audit_log/ # manage_system,也无机构/用户/角色管理位 → 仅方案管理页签 assert allowed_modules_for_user(lim_admin) == ["BACKEND_MANAGEMENT"] @pytest.mark.integration def test_limited_admin_without_role_follows_default(matrix_env): app, ids = matrix_env with app.app_context(): ops = db.session.get(Organization, ids["ops"]) # 未关联角色 → 回退默认角色(与 USER 语义一致) default_role = _make_role("受限管理员默认角色", is_default=True) lim_admin = _make_limited_admin(ops) _matrix_row(default_role.id, "SECRET", "OWN", can_view=True) doc = _make_document("受限管理员默认角色主案", org=ops) db.session.commit() context = _user_context(lim_admin) access = evaluate_plan_access(doc, context) assert access.allowed is True # 默认角色只有查看位:编辑拒绝 with pytest.raises(DocumentActionForbiddenError): require_plan_action(doc, context, AllowedAction.EDIT) @pytest.mark.integration def test_limited_admin_keeps_status_exemption_security_gate_and_attachment(matrix_env): app, ids = matrix_env with app.app_context(): ops = db.session.get(Organization, ids["ops"]) # 无角色且无默认角色:矩阵贡献为空 lim_admin = _make_limited_admin(ops) draft_doc = _make_document( "受限管理员草稿主案", org=ops, status="DRAFT" ) top_secret_doc = _make_document( "受限管理员绝密主案", org=ops, security="TOP_SECRET" ) att = _make_document( "受限管理员共享附件", org=ops, security="PUBLIC", document_type="ATTACHMENT", attachment_type="TRAINING_MATERIAL", ) db.session.commit() context = _user_context(lim_admin) # 草稿不按状态拒绝(状态豁免仍是管理员语义),矩阵/ACL 均无 → ACL draft_access = evaluate_plan_access(draft_doc, context) assert draft_access.allowed is False assert draft_access.reason == "ACL" # 密级前置任何角色不可绕过 assert evaluate_plan_access(top_secret_doc, context).reason == "SECURITY" # 共享附件维护仍是管理员级语义,不按矩阵限制 require_plan_action(att, context, AllowedAction.EDIT)