ctms/backend/tests/test_permission_security.py

"""安全测试：权限系统安全性验证

测试权限系统的安全性，包括：
- 权限检查遗漏检测
- 权限配置完整性
- 缓存失效场景
- 成员停用后的权限检查
- 权限更新后的立即生效
- 并发权限检查一致性
"""

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} 没有权限配置"