zhouchengcheng/ctms
1
1
Fork 0
Code Issues Pull Requests 2 Actions Packages Projects Releases Wiki Activity

权限监控与管理界面全面美化

Browse Source 
重新设计权限系统所有 UI 组件的视觉风格，统一配色、圆角、阴影和交互动效：
- 实时概览：统计卡片加图标和渐变色条，健康评分改为环形进度，告警改为卡片式
- 趋势分析：图表卡片加彩色图标标识，ECharts 配色升级为渐变面积填充
- 访问日志：指标卡片带图标，日志审计改为卡片式入口，IP排行前三高亮
- IP属地：工具栏重设计，排行列表前三渐变高亮，指标卡片统一新风格
- 系统级权限：从 el-table 改为自定义卡片列表，模块块独立圆角卡片
- 项目权限配置：空状态引导优化，成员表格加头像，工具栏加背景容器
- 角色概览卡片：加进度条可视化，hover 微动效
- 接口权限矩阵：工具栏分离布局，表格圆角包裹
- 角色管理抽屉：侧边栏选中态渐变，操作行 hover 高亮

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
This commit is contained in:
Cheng Zhou
2026-05-20 14:36:35 +08:00
parent e95eeed90e
commit 6cefa620e4
68 changed files with 6927 additions and 984 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/tests/conftest.py
+3
View File
@@ -83,6 +83,9 @@ async def _create_test_engine():
import app.models.monitoring_visit_issue
import app.models.site
import app.models.permission_template
import app.models.permission_access_log
import app.models.permission_metric_snapshot
import app.models.security_access_log
# Replace PostgreSQL UUID type with custom GUID type for SQLite
for table in Base.metadata.tables.values():
backend/tests/test_api_permissions.py
+112
View File
@@ -2,13 +2,19 @@
import pytest
import uuid
from pathlib import Path
import re
from sqlalchemy import delete
from sqlalchemy.ext.asyncio import AsyncSession
from app.core.api_permissions import API_ENDPOINT_PERMISSIONS, PROJECT_PERMISSION_ROLES
from app.core.project_permissions import role_has_api_permission
from app.core.project_permissions import get_api_endpoint_permissions, replace_api_endpoint_permissions
from app.core.permission_cache import PermissionCache, set_permission_cache
from app.core.permission_monitor import PermissionMonitor, set_permission_monitor
from app.models.api_endpoint_permission import ApiEndpointPermission
from app.models.study import Study
from app.schemas.member import StudyMemberCreate
@pytest.mark.asyncio
@@ -33,6 +39,112 @@ async def test_api_permission_check_allowed(db_session: AsyncSession):
assert result is True
def test_all_backend_permission_guards_are_configurable():
"""确保后端实际鉴权使用的 operation key 都能在权限管理中配置。"""
api_dir = Path(__file__).resolve().parents[1] / "app" / "api" / "v1"
used_keys: set[str] = set()
pattern = re.compile(r"require_api_permission\(\s*[\"']([^\"']+)[\"']")
for path in api_dir.rglob("*.py"):
used_keys.update(pattern.findall(path.read_text()))
assert used_keys
assert used_keys <= set(API_ENDPOINT_PERMISSIONS)
@pytest.mark.asyncio
async def test_default_matrix_covers_every_role_and_permission(db_session: AsyncSession):
"""逐一验证 6 个预设角色在每个权限上的默认矩阵。"""
study_id = uuid.uuid4()
matrix = await get_api_endpoint_permissions(db_session, study_id)
assert set(matrix) == set(PROJECT_PERMISSION_ROLES)
for role in PROJECT_PERMISSION_ROLES:
assert set(matrix[role]) == set(API_ENDPOINT_PERMISSIONS)
for endpoint_key, config in API_ENDPOINT_PERMISSIONS.items():
assert matrix[role][endpoint_key]["allowed"] is (role in config["default_roles"])
@pytest.mark.asyncio
async def test_full_permission_matrix_round_trips_to_backend_checks(db_session: AsyncSession):
"""逐一验证前端提交格式会落库，并被后端鉴权函数按相同结果读取。"""
cache = PermissionCache()
monitor = PermissionMonitor()
set_permission_cache(cache)
set_permission_monitor(monitor)
study_id = uuid.uuid4()
payload = {
role: {
endpoint_key: index % 2 == role_index % 2
for index, endpoint_key in enumerate(API_ENDPOINT_PERMISSIONS)
}
for role_index, role in enumerate(PROJECT_PERMISSION_ROLES)
}
matrix = await replace_api_endpoint_permissions(db_session, study_id, payload)
for role in PROJECT_PERMISSION_ROLES:
for endpoint_key, expected in payload[role].items():
assert matrix[role][endpoint_key]["allowed"] is expected
allowed = await role_has_api_permission(
db_session,
study_id,
role,
endpoint_key,
check_prerequisites=False,
)
assert allowed is expected
@pytest.mark.asyncio
async def test_custom_active_role_permissions_take_effect(db_session: AsyncSession):
"""验证项目自定义角色配置权限后能被后端鉴权读取。"""
study_id = uuid.uuid4()
db_session.add(
Study(
id=study_id,
code=f"CUSTOM-{study_id.hex[:8]}",
name="自定义角色权限测试",
status="ACTIVE",
is_locked=False,
visit_schedule=[],
active_roles=["DATA_MANAGER"],
)
)
await db_session.commit()
matrix = await get_api_endpoint_permissions(db_session, study_id)
assert "DATA_MANAGER" in matrix
assert matrix["DATA_MANAGER"]["subjects:read"]["allowed"] is False
await replace_api_endpoint_permissions(
db_session,
study_id,
{"DATA_MANAGER": {"subjects:read": True, "subjects:update": False}},
)
assert await role_has_api_permission(
db_session,
study_id,
"DATA_MANAGER",
"subjects:read",
check_prerequisites=False,
) is True
assert await role_has_api_permission(
db_session,
study_id,
"DATA_MANAGER",
"subjects:update",
check_prerequisites=False,
) is False
def test_admin_cannot_be_used_as_project_custom_role():
"""避免项目角色 ADMIN 触发后端系统管理员绕过逻辑。"""
with pytest.raises(ValueError):
StudyMemberCreate(user_id=uuid.uuid4(), role_in_study="ADMIN")
@pytest.mark.asyncio
async def test_api_permission_check_denied(db_session: AsyncSession):
"""测试接口级权限检查 - 拒绝"""
backend/tests/test_ip_location.py
+37
View File
@@ -0,0 +1,37 @@
"""单元测试：IP 属地解析服务"""
from app.services.ip_location import Ip2RegionResolver
class FakeSearcher:
def search(self, _ip: str) -> str:
return "中国|广东省|深圳市|电信|CN"
def test_ip_location_handles_special_addresses():
resolver = Ip2RegionResolver(db_path="/not-exists/ip2region.xdb")
assert resolver.lookup(None).location == "未知"
assert resolver.lookup("not-an-ip").location == "未知"
assert resolver.lookup("127.0.0.1").location == "本机"
assert resolver.lookup("192.168.1.1").location == "局域网"
def test_ip2region_result_parses_province_city_isp():
resolver = Ip2RegionResolver(db_path="/not-exists/ip2region.xdb")
resolver._searchers[4] = FakeSearcher()
result = resolver.lookup("8.8.8.8")
assert result.country == "中国"
assert result.province == "广东省"
assert result.city == "深圳市"
assert result.isp == "电信"
assert result.location == "中国 / 广东省 / 深圳市 / 电信"
def test_default_resolver_can_use_packaged_xdb():
resolver = Ip2RegionResolver()
result = resolver.lookup("8.8.8.8")
assert result.location not in {"公网", "未知"}
backend/tests/test_permission_monitoring_api.py
+250
View File
@@ -5,8 +5,19 @@
import pytest
from fastapi.testclient import TestClient
from sqlalchemy import text
from app.core.permission_monitor import get_permission_monitor, set_permission_monitor, PermissionMonitor
from app.api.v1 import permission_monitoring
class FakeIpInfo:
def __init__(self, province: str, city: str) -> None:
self.country = "中国"
self.province = province
self.city = city
self.isp = "电信"
self.location = f"中国 / {province} / {city} / 电信"
@pytest.mark.asyncio
@@ -205,3 +216,242 @@ async def test_permission_system_health_includes_metrics(client: TestClient, aut
assert "cache_stats" in data
assert "check_metrics" in data["metrics"]
assert "cache_metrics" in data["metrics"]
@pytest.mark.asyncio
async def test_ip_locations_counts_unique_users_per_location(db_session, monkeypatch):
"""IP 属地统计应按省市聚合访问次数、来源 IP 数和访问用户数。"""
await db_session.execute(text("DELETE FROM permission_access_logs"))
await db_session.commit()
study_id = "00000000-0000-0000-0000-000000000001"
user_a = "00000000-0000-0000-0000-000000000101"
user_b = "00000000-0000-0000-0000-000000000102"
await db_session.execute(
text(
"""
INSERT INTO studies (id, code, name, status, is_locked, visit_schedule, active_roles)
VALUES (:id, :code, :name, :status, :is_locked, :visit_schedule, :active_roles)
"""
),
{
"id": study_id,
"code": "IP-LOCATION-STUDY",
"name": "IP Location Study",
"status": "ACTIVE",
"is_locked": False,
"visit_schedule": "[]",
"active_roles": "[]",
},
)
for user_id, email in [(user_a, "user-a@example.com"), (user_b, "user-b@example.com")]:
await db_session.execute(
text(
"""
INSERT INTO users (id, email, password_hash, full_name, role, clinical_department, status)
VALUES (:id, :email, :password_hash, :full_name, :role, :clinical_department, :status)
"""
),
{
"id": user_id,
"email": email,
"password_hash": "hash",
"full_name": email,
"role": "PM",
"clinical_department": "临床运营",
"status": "ACTIVE",
},
)
rows = [
(study_id, user_a, "10.1.1.1", True),
(study_id, user_a, "10.1.1.1", False),
(study_id, user_b, "10.1.1.2", True),
]
for study, user, ip_address, allowed in rows:
await db_session.execute(
text(
"""
INSERT INTO permission_access_logs
(id, study_id, user_id, endpoint_key, role, allowed, elapsed_ms, ip_address, created_at)
VALUES
(lower(hex(randomblob(4))) || '-' || lower(hex(randomblob(2))) || '-4' ||
substr(lower(hex(randomblob(2))),2) || '-' ||
substr('89ab', abs(random()) % 4 + 1, 1) ||
substr(lower(hex(randomblob(2))),2) || '-' || lower(hex(randomblob(6))),
:study_id, :user_id, :endpoint_key, :role, :allowed, :elapsed_ms, :ip_address, CURRENT_TIMESTAMP)
"""
),
{
"study_id": study,
"user_id": user,
"endpoint_key": "admin.permissions.read",
"role": "PM",
"allowed": allowed,
"elapsed_ms": 3.2,
"ip_address": ip_address,
},
)
await db_session.commit()
monkeypatch.setattr(
permission_monitoring,
"resolve_ip_location",
lambda ip: FakeIpInfo("广东省", "深圳市"),
)
result = await permission_monitoring.get_ip_locations(db=db_session, _=object(), days=7, limit=10)
assert result["items"][0]["province"] == "广东省"
assert result["items"][0]["total_count"] == 3
assert result["items"][0]["unique_ip_count"] == 2
assert result["items"][0]["unique_user_count"] == 2
assert result["summary"]["total_count"] == 3
assert result["summary"]["unique_ip_count"] == 2
assert result["summary"]["unique_user_count"] == 2
@pytest.mark.asyncio
async def test_access_logs_include_user_behavior_summary(db_session):
"""访问日志应返回用户行为审计汇总和用户排行。"""
await db_session.execute(text("DELETE FROM permission_access_logs"))
await db_session.commit()
study_id = "00000000-0000-0000-0000-000000000201"
user_a = "00000000-0000-0000-0000-000000000301"
user_b = "00000000-0000-0000-0000-000000000302"
await db_session.execute(
text(
"""
INSERT INTO studies (id, code, name, status, is_locked, visit_schedule, active_roles)
VALUES (:id, :code, :name, :status, :is_locked, :visit_schedule, :active_roles)
"""
),
{
"id": study_id,
"code": "ACCESS-LOG-STUDY",
"name": "Access Log Study",
"status": "ACTIVE",
"is_locked": False,
"visit_schedule": "[]",
"active_roles": "[]",
},
)
for user_id, email, name, role in [
(user_a, "audit-a@example.com", "审计用户A", "PM"),
(user_b, "audit-b@example.com", "审计用户B", "CRA"),
]:
await db_session.execute(
text(
"""
INSERT INTO users (id, email, password_hash, full_name, role, clinical_department, status)
VALUES (:id, :email, :password_hash, :full_name, :role, :clinical_department, :status)
"""
),
{
"id": user_id,
"email": email,
"password_hash": "hash",
"full_name": name,
"role": role,
"clinical_department": "临床运营",
"status": "ACTIVE",
},
)
rows = [
(study_id, user_a, "admin.permissions.read", "PM", True, 3.0, "10.1.1.1"),
(study_id, user_a, "admin.users.delete", "PM", False, 9.0, "10.1.1.2"),
(study_id, user_b, "admin.permissions.read", "CRA", True, 6.0, "10.1.1.3"),
]
for study, user, endpoint, role, allowed, elapsed_ms, ip_address in rows:
await db_session.execute(
text(
"""
INSERT INTO permission_access_logs
(id, study_id, user_id, endpoint_key, role, allowed, elapsed_ms, ip_address, created_at)
VALUES
(lower(hex(randomblob(4))) || '-' || lower(hex(randomblob(2))) || '-4' ||
substr(lower(hex(randomblob(2))),2) || '-' ||
substr('89ab', abs(random()) % 4 + 1, 1) ||
substr(lower(hex(randomblob(2))),2) || '-' || lower(hex(randomblob(6))),
:study_id, :user_id, :endpoint_key, :role, :allowed, :elapsed_ms, :ip_address, CURRENT_TIMESTAMP)
"""
),
{
"study_id": study,
"user_id": user,
"endpoint_key": endpoint,
"role": role,
"allowed": allowed,
"elapsed_ms": elapsed_ms,
"ip_address": ip_address,
},
)
await db_session.commit()
result = await permission_monitoring.get_access_logs(
db=db_session,
_=object(),
study_id=None,
user_id=None,
endpoint_key=None,
role=None,
allowed=None,
start_time=None,
end_time=None,
page=1,
page_size=50,
)
assert result["summary"]["total_count"] == 3
assert result["summary"]["unique_user_count"] == 2
assert result["summary"]["denied_count"] == 1
assert result["summary"]["avg_elapsed_ms"] == 6.0
assert result["user_stats"][0]["user_name"] == "审计用户A"
assert result["user_stats"][0]["total_count"] == 1
assert result["user_stats"][0]["denied_count"] in {0, 1}
assert result["user_stats"][0]["unique_ip_count"] == 1
assert result["user_stats"][0]["sample_ip_address"] in {"10.1.1.1", "10.1.1.2"}
user_a_ips = {
stat["sample_ip_address"]
for stat in result["user_stats"]
if stat["user_name"] == "审计用户A"
}
assert user_a_ips == {"10.1.1.1", "10.1.1.2"}
@pytest.mark.asyncio
async def test_security_access_logs_include_anonymous_ip_attempts(db_session):
"""安全访问日志应覆盖未登录或未知账号的底层访问尝试。"""
await db_session.execute(text("DELETE FROM security_access_logs"))
await db_session.execute(
text(
"""
INSERT INTO security_access_logs
(id, method, path, status_code, elapsed_ms, client_ip, user_agent, auth_status, user_identifier, created_at)
VALUES
('00000000-0000-4000-8000-000000000501', 'POST', '/api/v1/auth/login', 401, 12.5,
'203.0.113.10', 'attack-bot/1.0', 'ANONYMOUS', NULL, CURRENT_TIMESTAMP)
"""
)
)
await db_session.commit()
result = await permission_monitoring.get_security_access_logs(
db=db_session,
_=object(),
status_min=400,
auth_status=None,
page=1,
page_size=20,
)
assert result["summary"]["total_count"] == 1
assert result["summary"]["anonymous_count"] == 1
assert result["summary"]["error_count"] == 1
assert result["items"][0]["client_ip"] == "203.0.113.10"
assert result["items"][0]["account_label"] == "未知账号"
assert result["items"][0]["status_code"] == 401