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

修复(权限管理)：统一 PM 系统导航与权限校验

Browse Source
This commit is contained in:
Cheng Zhou
2026-07-10 22:12:59 +08:00
parent f11a5c84d9
commit f2fa10c178
47 changed files with 4071 additions and 1555 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/tests/test_api_permissions.py
+10
View File
@@ -565,6 +565,16 @@ def test_project_permission_config_is_system_level_permission():
assert "PM" in SYSTEM_PERMISSIONS["system:permissions:project_config"]["roles"]
def test_pm_system_management_navigation_matches_backend_permission_contract():
"""PM 系统管理导航中的审计日志和权限管理应与后端角色权限保持一致。"""
assert "PM" in API_ENDPOINT_PERMISSIONS["audit_logs:read"]["default_roles"]
assert "PM" in SYSTEM_PERMISSIONS["system:permissions:read"]["roles"]
assert "PM" in SYSTEM_PERMISSIONS["system:permissions:project_config"]["roles"]
audit_route = Path(__file__).resolve().parents[1] / "app" / "api" / "v1" / "audit_logs.py"
assert 'require_api_permission("audit_logs:read")' in audit_route.read_text()
def test_project_api_permission_routes_use_system_project_config_permission():
"""项目权限矩阵 API 应明确依赖 system:permissions:project_config。"""
route_path = Path(__file__).resolve().parents[1] / "app" / "api" / "v1" / "api_permissions.py"
backend/tests/test_permission_monitoring_api.py
+87 -5
View File
@@ -69,7 +69,19 @@ async def test_system_permission_monitoring_definitions_are_admin_only():
assert all("PM" not in item["roles"] for item in monitoring_items)
async def _seed_permission_log(db_session, study_id: uuid.UUID, user_id: uuid.UUID, *, allowed: bool, elapsed_ms: float) -> None:
async def _seed_permission_log(
db_session,
study_id: uuid.UUID,
user_id: uuid.UUID,
*,
allowed: bool,
elapsed_ms: float,
endpoint_key: str = "admin.permissions.read",
role: str = "PM",
client_type: str | None = None,
client_version: str | None = None,
client_platform: str | None = None,
) -> None:
study_exists = (
await db_session.execute(text("SELECT id FROM studies WHERE id = :id"), {"id": str(study_id)})
).scalar_one_or_none()
@@ -119,21 +131,24 @@ async def _seed_permission_log(db_session, study_id: uuid.UUID, user_id: uuid.UU
"""
INSERT INTO permission_access_logs
(id, study_id, user_id, endpoint_key, role, allowed, elapsed_ms, ip_address,
request_snapshot, created_at)
client_type, client_version, client_platform, request_snapshot, created_at)
VALUES
(:id, :study_id, :user_id, :endpoint_key, :role, :allowed, :elapsed_ms, :ip_address,
:request_snapshot, CURRENT_TIMESTAMP)
:client_type, :client_version, :client_platform, :request_snapshot, CURRENT_TIMESTAMP)
"""
),
{
"id": str(uuid.uuid4()),
"study_id": str(study_id),
"user_id": str(user_id),
"endpoint_key": "admin.permissions.read",
"role": "PM",
"endpoint_key": endpoint_key,
"role": role,
"allowed": allowed,
"elapsed_ms": elapsed_ms,
"ip_address": "127.0.0.1",
"client_type": client_type,
"client_version": client_version,
"client_platform": client_platform,
"request_snapshot": None,
},
)
@@ -159,6 +174,53 @@ async def test_get_permission_metrics(db_session):
assert data["check_metrics"]["denied_checks"] == 1
@pytest.mark.asyncio
async def test_get_trends_returns_dense_summary_comparison_and_attribution(db_session):
baseline = await permission_monitoring.get_trends(db=db_session, _=AdminUserStub(), period="24h")
study_id = uuid.uuid4()
user_id = uuid.uuid4()
await _seed_permission_log(
db_session,
study_id,
user_id,
allowed=True,
elapsed_ms=5,
endpoint_key="subjects.read",
client_type="web",
client_version="0.1.0",
client_platform="macos",
)
await _seed_permission_log(
db_session,
study_id,
user_id,
allowed=False,
elapsed_ms=80,
endpoint_key="subjects.export",
client_type="web",
client_version="0.1.0",
client_platform="macos",
)
data = await permission_monitoring.get_trends(db=db_session, _=AdminUserStub(), period="24h")
assert data["period"] == "24h"
assert data["range"]["bucket_seconds"] == 3600
assert len(data["data_points"]) == 24
assert data["data_points"][-1]["sample_state"] == "partial"
assert data["summary"]["total_checks"] == baseline["summary"]["total_checks"] + 2
assert data["summary"]["denied_checks"] == baseline["summary"]["denied_checks"] + 1
assert data["summary"]["slow_check_count"] == baseline["summary"]["slow_check_count"] + 1
assert data["summary"]["active_study_count"] == baseline["summary"]["active_study_count"] + 1
assert data["summary"]["active_user_count"] == baseline["summary"]["active_user_count"] + 1
assert data["summary"]["active_endpoint_count"] >= 2
assert data["summary"]["max_elapsed_ms"] >= data["summary"]["p95_elapsed_ms"]
assert data["previous_summary"]["total_checks"] == 0
assert any(item["endpoint_key"] == "subjects.export" for item in data["attribution"]["top_denied"])
assert data["attribution"]["top_slow"][0]["endpoint_key"] == "subjects.export"
assert any(item["client_type"] == "web" for item in data["attribution"]["client_breakdown"])
@pytest.mark.asyncio
async def test_get_cache_statistics(db_session):
"""测试获取缓存统计"""
@@ -1206,6 +1268,26 @@ async def test_access_logs_include_security_events_for_admin(db_session, monkeyp
assert items_by_type["permission"]["request_snapshot"]["path"] == "/api/v1/projects/overview"
assert items_by_type["security"]["request_snapshot"]["headers"]["authorization"] == "[redacted]"
permission_only = await permission_monitoring.get_access_logs(
db=db_session,
_=AdminUserStub(),
study_id=None,
user_id=None,
endpoint_key=None,
role=None,
allowed=None,
start_time=None,
end_time=None,
client_ip=None,
client_type=None,
event_type="permission",
keyword=None,
page=1,
page_size=50,
)
assert permission_only["total"] == 1
assert permission_only["items"][0]["event_type"] == "permission"
@pytest.mark.asyncio
async def test_access_logs_ip_ranking_aggregates_by_ip_total(db_session):
backend/tests/test_user_login_sessions.py
+42
View File
@@ -4,6 +4,7 @@ import uuid
import pytest
from app.core.config import settings
from app.crud import user as user_crud
from app.models.user import User, UserStatus
from app.models.user_login_session import UserLoginSession
from app.services.user_login_sessions import get_login_summaries, session_id_from_payload
@@ -59,3 +60,44 @@ async def test_login_summary_marks_recent_unended_sessions_online(db_session, mo
assert summaries[user.id].status == "ONLINE"
assert summaries[user.id].active_session_count == 1
assert summaries[user.id].client_type == "desktop"
@pytest.mark.asyncio
async def test_user_list_filters_online_and_offline_accounts(db_session, monkeypatch):
now = datetime.now(timezone.utc)
monkeypatch.setattr(settings, "USER_SESSION_ONLINE_SECONDS", 300)
online_user = User(
id=uuid.uuid4(),
email="online-filter@example.com",
password_hash="hash",
full_name="Online Filter",
clinical_department="IT",
status=UserStatus.ACTIVE,
)
offline_user = User(
id=uuid.uuid4(),
email="offline-filter@example.com",
password_hash="hash",
full_name="Offline Filter",
clinical_department="IT",
status=UserStatus.ACTIVE,
)
db_session.add_all([online_user, offline_user])
db_session.add(
UserLoginSession(
id=uuid.uuid4(),
user_id=online_user.id,
client_type="web",
login_at=now - timedelta(minutes=1),
last_seen_at=now - timedelta(seconds=10),
)
)
await db_session.commit()
online = await user_crud.list_users(db_session, login_status="ONLINE")
offline = await user_crud.list_users(db_session, login_status="OFFLINE")
assert online_user.id in {user.id for user in online}
assert offline_user.id not in {user.id for user in online}
assert offline_user.id in {user.id for user in offline}
assert await user_crud.count_users(db_session, login_status="ONLINE") == len(online)