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

完善权限监控与安全中心

Browse Source
This commit is contained in:
Cheng Zhou
2026-06-17 17:21:48 +08:00
parent 1886765db8
commit 061792c73f
15 changed files with 2837 additions and 160 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/tests/test_permission_monitoring_api.py
+335 -4
View File
@@ -1,5 +1,6 @@
"""监控API测试：权限系统监控API端点验证。"""
import inspect
import uuid
import pytest
@@ -12,12 +13,12 @@ from app.api.v1.system_permissions import list_system_permissions
class FakeIpInfo:
def __init__(self, province: str, city: str, isp: str = "电信") -> None:
self.country = "中国"
def __init__(self, province: str, city: str, isp: str = "电信", country: str = "中国") -> None:
self.country = country
self.province = province
self.city = city
self.isp = isp
self.location = f"中国 / {province} / {city} / {isp}"
self.location = " / ".join(part for part in [country, province, city, isp] if part)
class AdminUserStub:
@@ -526,6 +527,117 @@ async def test_ip_locations_keeps_private_network_location_label(db_session, mon
assert result["items"][0]["city"] == ""
@pytest.mark.asyncio
async def test_ip_locations_includes_abnormal_security_ips(db_session, monkeypatch):
"""来源分析应同时统计安全事件中的异常来源 IP。"""
await db_session.execute(text("DELETE FROM permission_access_logs"))
await db_session.execute(text("DELETE FROM security_access_logs"))
await db_session.commit()
study_id = "00000000-0000-0000-0000-000000000801"
user_id = "00000000-0000-0000-0000-000000000901"
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-SECURITY-STUDY",
"name": "IP Location Security Study",
"status": "ACTIVE",
"is_locked": False,
"visit_schedule": "[]",
"active_roles": "[]",
},
)
await db_session.execute(
text(
"""
INSERT INTO users (id, email, password_hash, full_name, clinical_department, is_admin, status)
VALUES (:id, :email, :password_hash, :full_name, :clinical_department, :is_admin, :status)
"""
),
{
"id": user_id,
"email": "source-analysis@example.com",
"password_hash": "hash",
"full_name": "来源分析用户",
"clinical_department": "临床运营",
"is_admin": False,
"status": "ACTIVE",
},
)
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
(:id, :study_id, :user_id, :endpoint_key, :role, :allowed, :elapsed_ms, :ip_address, CURRENT_TIMESTAMP)
"""
),
{
"id": str(uuid.uuid4()),
"study_id": study_id,
"user_id": user_id,
"endpoint_key": "admin.permissions.read",
"role": "PM",
"allowed": True,
"elapsed_ms": 3.2,
"ip_address": "10.3.1.1",
},
)
for ip_address in ["8.8.8.8", "1.1.1.1"]:
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
(:id, :method, :path, :status_code, :elapsed_ms, :client_ip, :user_agent, :auth_status, :user_identifier, CURRENT_TIMESTAMP)
"""
),
{
"id": str(uuid.uuid4()),
"method": "GET",
"path": "/api/v1/admin",
"status_code": 403,
"elapsed_ms": 8.5,
"client_ip": ip_address,
"user_agent": "pytest",
"auth_status": "ANONYMOUS",
"user_identifier": None,
},
)
await db_session.commit()
ip_info_by_address = {
"10.3.1.1": FakeIpInfo("", "", "", "局域网"),
"8.8.8.8": FakeIpInfo("California", "Mountain View", "Google", "United States"),
"1.1.1.1": FakeIpInfo("Queensland", "Brisbane", "Cloudflare", "Australia"),
}
monkeypatch.setattr(
permission_monitoring,
"resolve_ip_location",
lambda ip: ip_info_by_address[ip],
)
result = await permission_monitoring.get_ip_locations(db=db_session, _=AdminUserStub(), days=7, limit=10)
locations = {item["country"]: item for item in result["items"]}
assert result["summary"]["total_count"] == 3
assert result["summary"]["denied_count"] == 2
assert result["summary"]["unique_ip_count"] == 3
assert result["summary"]["unique_user_count"] == 1
assert locations["United States"]["total_count"] == 1
assert locations["United States"]["denied_count"] == 1
assert locations["Australia"]["total_count"] == 1
@pytest.mark.asyncio
async def test_access_logs_include_user_behavior_summary(db_session):
"""访问日志应返回用户行为审计汇总和用户排行。"""
@@ -637,10 +749,121 @@ async def test_access_logs_include_user_behavior_summary(db_session):
assert user_a_ips == {"10.1.1.1", "10.1.1.2"}
def test_access_logs_user_stats_query_avoids_postgresql_uuid_max():
"""访问日志用户统计不能对 UUID 字段使用 max，PostgreSQL 不支持 max(uuid)。"""
source = inspect.getsource(permission_monitoring.get_access_logs)
assert "func.max(PermissionAccessLog.user_id)" not in source
@pytest.mark.asyncio
async def test_security_access_logs_include_anonymous_ip_attempts(db_session):
async def test_access_logs_ip_ranking_aggregates_by_ip_total(db_session):
"""IP 访问排行应按同一 IP 的整体访问量聚合排序，而不是按用户/IP/角色拆分。"""
await db_session.execute(text("DELETE FROM permission_access_logs"))
await db_session.commit()
study_id = "00000000-0000-0000-0000-000000001201"
user_a = "00000000-0000-0000-0000-000000001301"
user_b = "00000000-0000-0000-0000-000000001302"
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-IP-RANK-STUDY",
"name": "Access IP Rank Study",
"status": "ACTIVE",
"is_locked": False,
"visit_schedule": "[]",
"active_roles": "[]",
},
)
for user_id, email, name in [
(user_a, "ip-rank-a@example.com", "IP排行用户A"),
(user_b, "ip-rank-b@example.com", "IP排行用户B"),
]:
await db_session.execute(
text(
"""
INSERT INTO users (id, email, password_hash, full_name, clinical_department, is_admin, status)
VALUES (:id, :email, :password_hash, :full_name, :clinical_department, :is_admin, :status)
"""
),
{
"id": user_id,
"email": email,
"password_hash": "hash",
"full_name": name,
"clinical_department": "临床运营",
"is_admin": False,
"status": "ACTIVE",
},
)
rows = [
(user_a, "183.230.169.20", True),
(user_a, "183.230.169.20", True),
(user_b, "183.230.169.20", False),
(user_a, "45.148.10.95", False),
(user_b, "52.53.218.145", False),
]
for 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
(:id, :study_id, :user_id, :endpoint_key, :role, :allowed, :elapsed_ms, :ip_address, CURRENT_TIMESTAMP)
"""
),
{
"id": str(uuid.uuid4()),
"study_id": study_id,
"user_id": user,
"endpoint_key": "admin.permissions.read",
"role": "CRA",
"allowed": allowed,
"elapsed_ms": 4.0,
"ip_address": ip_address,
},
)
await db_session.commit()
result = 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,
page=1,
page_size=50,
)
assert result["user_stats"][0]["sample_ip_address"] == "183.230.169.20"
assert result["user_stats"][0]["total_count"] == 3
assert result["user_stats"][0]["denied_count"] == 1
assert {stat["sample_ip_address"] for stat in result["user_stats"][1:]} == {"52.53.218.145", "45.148.10.95"}
@pytest.mark.asyncio
async def test_security_access_logs_include_anonymous_ip_attempts(db_session, monkeypatch):
"""安全访问日志应覆盖未登录或未知账号的底层访问尝试。"""
await db_session.execute(text("DELETE FROM security_access_logs"))
monkeypatch.setattr(
permission_monitoring,
"resolve_ip_location",
lambda ip: FakeIpInfo("上海市", "上海市", "联通"),
)
await db_session.execute(
text(
"""
@@ -667,5 +890,113 @@ async def test_security_access_logs_include_anonymous_ip_attempts(db_session):
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]["ip_location"] == "中国 / 上海市 / 上海市 / 联通"
assert result["items"][0]["ip_country"] == "中国"
assert result["items"][0]["ip_province"] == "上海市"
assert result["items"][0]["ip_city"] == "上海市"
assert result["items"][0]["ip_isp"] == "联通"
assert result["items"][0]["account_label"] == "未知账号"
assert result["items"][0]["status_code"] == 401
@pytest.mark.asyncio
async def test_security_access_logs_classify_sensitive_path_probe(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-000000000502', 'GET', '/api/.git/config', 404, 0.7,
'198.51.100.20', 'probe-bot/1.0', 'ANONYMOUS', NULL, CURRENT_TIMESTAMP)
"""
)
)
await db_session.commit()
result = await permission_monitoring.get_security_access_logs(
db=db_session,
_=AdminUserStub(),
status_min=400,
auth_status=None,
page=1,
page_size=20,
)
assert result["items"][0]["category"] == "PROBE"
assert result["items"][0]["severity"] == "CRITICAL"
@pytest.mark.asyncio
async def test_security_access_logs_classify_non_china_ip_as_abnormal(db_session, monkeypatch):
"""安全事件明细应把非中国公网 IP 归类为异常 IP。"""
await db_session.execute(text("DELETE FROM security_access_logs"))
monkeypatch.setattr(
permission_monitoring,
"resolve_ip_location",
lambda ip: FakeIpInfo("California", "San Jose", "Google", "United States"),
)
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-000000000504', 'GET', '/api/v1/studies/', 401, 8.4,
'52.53.218.145', 'foreign-client/1.0', 'INVALID_TOKEN', NULL, CURRENT_TIMESTAMP)
"""
)
)
await db_session.commit()
result = await permission_monitoring.get_security_access_logs(
db=db_session,
_=AdminUserStub(),
status_min=400,
auth_status=None,
page=1,
page_size=20,
)
item = result["items"][0]
assert item["category"] == "ABNORMAL_IP"
assert item["severity"] == "HIGH"
assert item["ip_location"] == "United States / California / San Jose / Google"
assert item["ip_country"] == "United States"
@pytest.mark.asyncio
async def test_security_access_logs_resolve_public_ip_with_packaged_ip2region(db_session):
"""安全中心公网 IP 应复用 ip2region 解析结果，避免前端只能显示未知位置。"""
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-000000000503', 'GET', '/api/v1/auth/login', 404, 1.2,
'52.53.218.145', 'probe-bot/1.0', 'ANONYMOUS', NULL, CURRENT_TIMESTAMP)
"""
)
)
await db_session.commit()
result = await permission_monitoring.get_security_access_logs(
db=db_session,
_=AdminUserStub(),
status_min=400,
auth_status=None,
page=1,
page_size=20,
)
item = result["items"][0]
assert item["client_ip"] == "52.53.218.145"
assert item["ip_location"] not in {"", "公网", "未知"}
assert item["ip_country"] == "United States"
assert item["ip_province"] == "California"
assert item["ip_city"] == "San Jose"
assert item["category"] == "ABNORMAL_IP"