"""安全测试：权限系统安全性验证 测试权限系统的安全性，包括： - 权限检查遗漏检测 - 权限配置完整性 - 缓存失效场景 - 成员停用后的权限检查 - 权限更新后的立即生效 - 并发权限检查一致性 """ import uuid import pytest from sqlalchemy.ext.asyncio import AsyncSession from app.core.api_permissions import API_ENDPOINT_PERMISSIONS from app.core.permission_cache import get_permission_cache from app.core.project_permissions import ( role_has_api_permission, role_has_project_permission, replace_project_role_permissions, replace_api_endpoint_permissions, ) @pytest.mark.asyncio async def test_all_endpoints_have_permission_checks(): """测试所有端点都配置了权限检查 这个测试验证 API_ENDPOINT_PERMISSIONS 中的所有端点都有完整的权限配置。 """ for endpoint_key, config in API_ENDPOINT_PERMISSIONS.items(): # 验证必需字段 assert "module" in config, f"Missing 'module' in {endpoint_key}" assert "action" in config, f"Missing 'action' in {endpoint_key}" assert "description" in config, f"Missing 'description' in {endpoint_key}" assert "default_roles" in config, f"Missing 'default_roles' in {endpoint_key}" # 验证字段值 assert config["action"] in ["read", "write"], f"Invalid action in {endpoint_key}" assert isinstance(config["default_roles"], list), f"default_roles must be list in {endpoint_key}" assert len(config["default_roles"]) > 0, f"default_roles must not be empty in {endpoint_key}" @pytest.mark.asyncio async def test_all_endpoints_in_api_permissions(): """测试所有端点都在 API_ENDPOINT_PERMISSIONS 中定义""" # 这个测试确保没有端点被遗漏 assert len(API_ENDPOINT_PERMISSIONS) > 0, "API_ENDPOINT_PERMISSIONS is empty" assert len(API_ENDPOINT_PERMISSIONS) >= 94, "Missing endpoints in API_ENDPOINT_PERMISSIONS" @pytest.mark.asyncio async def test_all_endpoints_have_default_roles(): """测试所有端点都有默认角色配置""" valid_roles = {"PM", "CRA", "PV", "MEDICAL_REVIEW", "IMP", "QA"} for endpoint_key, config in API_ENDPOINT_PERMISSIONS.items(): default_roles = config.get("default_roles", []) assert len(default_roles) > 0, f"No default roles for {endpoint_key}" for role in default_roles: assert role in valid_roles, f"Invalid role {role} in {endpoint_key}" @pytest.mark.asyncio async def test_inactive_member_permission_denied(db_session: AsyncSession, study_id: uuid.UUID): """测试成员停用后的权限检查 当成员被停用时，应该拒绝其权限请求。 """ # 这个测试需要创建一个停用的成员 # 然后验证权限检查返回 False # 具体实现取决于成员模型的设计 # 暂时跳过，等待成员管理的完整实现 pass @pytest.mark.asyncio async def test_permission_update_immediate_effect(db_session: AsyncSession, study_id: uuid.UUID): """测试权限更新后的立即生效 当权限被更新时，新的权限应该立即生效。 """ # 获取初始权限 initial_allowed = await role_has_project_permission(db_session, study_id, "CRA", "subjects", "write") # 更新权限：允许 CRA 写入 subjects new_permissions = { "PM": {"subjects": {"read": True, "write": True}}, "CRA": {"subjects": {"read": True, "write": True}}, } await replace_project_role_permissions(db_session, study_id, new_permissions) # 验证权限立即生效 updated_allowed = await role_has_project_permission(db_session, study_id, "CRA", "subjects", "write") assert updated_allowed is True, "权限更新未立即生效" @pytest.mark.asyncio async def test_cache_invalidation_on_permission_update(db_session: AsyncSession, study_id: uuid.UUID): """测试权限更新时的缓存失效 当权限被更新时，相关的缓存应该被失效。 """ cache = get_permission_cache() cache.clear_all() # 填充缓存 await role_has_project_permission(db_session, study_id, "PM", "subjects", "read") assert cache.get_cache_stats()["project_permissions_count"] == 1 # 更新权限 new_permissions = { "PM": {"subjects": {"read": True, "write": True}}, } await replace_project_role_permissions(db_session, study_id, new_permissions) # 验证缓存已失效 assert cache.get_cache_stats()["project_permissions_count"] == 0 @pytest.mark.asyncio async def test_cache_invalidation_on_member_update(db_session: AsyncSession, study_id: uuid.UUID, user_id: uuid.UUID): """测试成员更新时的缓存失效 当成员角色被更新时，相关的缓存应该被失效。 """ cache = get_permission_cache() cache.clear_all() # 填充缓存 await role_has_project_permission(db_session, study_id, "PM", "subjects", "read") initial_stats = cache.get_cache_stats() # 失效成员角色缓存 cache.invalidate_member_role(study_id, user_id) # 验证成员角色缓存已失效 # （项目权限缓存应该保留） assert cache.get_cache_stats()["member_role_count"] == 0 @pytest.mark.asyncio async def test_concurrent_permission_checks_consistency(db_session: AsyncSession, study_id: uuid.UUID): """测试并发权限检查的一致性 多个并发请求检查权限时，结果应该一致。 """ import asyncio async def check_permission(): return await role_has_project_permission(db_session, study_id, "PM", "subjects", "read") # 并发执行权限检查 tasks = [check_permission() for _ in range(10)] results = await asyncio.gather(*tasks) # 所有结果应该相同 assert all(r == results[0] for r in results), "并发权限检查结果不一致" @pytest.mark.asyncio async def test_api_permission_priority_over_module(db_session: AsyncSession, study_id: uuid.UUID): """测试接口级权限优先于模块级权限 当同时配置了接口级和模块级权限时，接口级权限应该优先。 """ # 设置模块级权限：允许 CRA 读取 subjects module_permissions = { "PM": {"subjects": {"read": True, "write": True}}, "CRA": {"subjects": {"read": True, "write": False}}, } await replace_project_role_permissions(db_session, study_id, module_permissions) # 设置接口级权限：拒绝 CRA 读取 subjects api_permissions = { "PM": {"GET:/subjects": True}, "CRA": {"GET:/subjects": False}, } await replace_api_endpoint_permissions(db_session, study_id, api_permissions) # 验证接口级权限优先 allowed = await role_has_api_permission(db_session, study_id, "CRA", "GET:/subjects") assert allowed is False, "接口级权限未优先于模块级权限" @pytest.mark.asyncio async def test_admin_role_always_allowed(db_session: AsyncSession, study_id: uuid.UUID): """测试 ADMIN 角色总是被允许 ADMIN 角色应该在所有权限检查中都被允许。 """ # 即使没有配置权限，ADMIN 也应该被允许 allowed = await role_has_project_permission(db_session, study_id, "ADMIN", "subjects", "write") assert allowed is True, "ADMIN 角色权限检查失败" # API 权限检查也应该允许 ADMIN api_allowed = await role_has_api_permission(db_session, study_id, "ADMIN", "POST:/subjects") assert api_allowed is True, "ADMIN 角色 API 权限检查失败" @pytest.mark.asyncio async def test_none_role_permission_denied(db_session: AsyncSession, study_id: uuid.UUID): """测试 None 角色被拒绝 当角色为 None 时，权限检查应该返回 False。 """ allowed = await role_has_project_permission(db_session, study_id, None, "subjects", "read") assert allowed is False, "None 角色权限检查失败" @pytest.mark.asyncio async def test_unknown_endpoint_permission_denied(db_session: AsyncSession, study_id: uuid.UUID): """测试未知端点被拒绝 当端点不存在时，权限检查应该返回 False。 """ allowed = await role_has_api_permission(db_session, study_id, "PM", "GET:/unknown-endpoint") assert allowed is False, "未知端点权限检查失败" @pytest.mark.asyncio async def test_permission_matrix_consistency(db_session: AsyncSession, study_id: uuid.UUID): """测试权限矩阵的一致性 权限矩阵应该包含所有角色和端点的权限信息。 """ from app.core.project_permissions import get_project_role_permissions permissions = await get_project_role_permissions(db_session, study_id) # 验证矩阵结构 assert isinstance(permissions, dict), "权限矩阵格式错误" # 验证所有角色都存在 for role in ["PM", "CRA", "PV", "MEDICAL_REVIEW", "IMP", "QA"]: assert role in permissions, f"缺少角色 {role}" # 验证每个角色都有权限配置 for role, role_permissions in permissions.items(): assert isinstance(role_permissions, dict), f"角色 {role} 的权限配置格式错误" assert len(role_permissions) > 0, f"角色 {role} 没有权限配置"