diff --git a/backend/app/modules/admin/api/announcement.py b/backend/app/modules/admin/api/announcement.py index f65cf13a..46b26136 100644 --- a/backend/app/modules/admin/api/announcement.py +++ b/backend/app/modules/admin/api/announcement.py @@ -7,7 +7,7 @@ - `frontend_router`(prefix=/announcements,在 app/__init__.py 以 /api/announcements 挂载) 鉴权: - - 后台:require_permission("frontend:admin:announcement:manage") + - 后台:require_permission(PERM_ANNOUNCEMENT),码定义在 permission_service - 前台:仅要求登录(App 全量在 AuthGuard 内,无匿名场景) Status 用字符串列(draft/published/offline),见 app/models/announcement.py。 @@ -27,13 +27,16 @@ from app.models.announcement import Announcement, AnnouncementStatus from app.modules.admin.api.auth import require_permission from app.services.identity_service import IdentityService +from app.services.permission_service import PERM_ANNOUNCEMENT logger = logging.getLogger(__name__) router = APIRouter(prefix="/announcements", tags=["admin-announcements"]) frontend_router = APIRouter(prefix="/announcements", tags=["announcements"]) -PERM_CODE = "frontend:admin:announcement:manage" +# 与 permission_service 的派生表共用同一个常量:全局角色「超级管理员」默认拿到它, +# 其他角色可在权限管理里勾选。两处若各写一份字符串,迟早漂移。 +PERM_CODE = PERM_ANNOUNCEMENT VALID_STATUS = ( AnnouncementStatus.DRAFT, @@ -50,7 +53,13 @@ def ensure_announcement_permission() -> None: - """确保权限码已注册到 permissions 表(幂等,已存在自动跳过)。""" + """确保权限码已注册到 permissions 表(幂等,已存在自动跳过)。 + + 只在「权限管理」列表的注册块(permissions.get_all_permissions)里调用: + 该码要在权限管理页可勾选,就必须在**读到权限列表时**先落库。切勿在下面各接口 + 函数体内调用——require_permission 是 FastAPI 依赖,先于函数体执行,没权限的 + 账号会先被 403 挡住,永远走不到注册那行(生产上就是这样自锁的)。 + """ IdentityService.add_permission( permission_id="perm_frontend_admin_announcement_manage", code=PERM_CODE, @@ -153,7 +162,6 @@ async def admin_list_announcements( db: AsyncSession = Depends(get_db), _: Dict[str, Any] = require_permission(PERM_CODE), ): - ensure_announcement_permission() conds = [] if status: conds.append(Announcement.status == _validate_status(status)) @@ -180,7 +188,6 @@ async def create_announcement( db: AsyncSession = Depends(get_db), current_user: Dict[str, Any] = require_permission(PERM_CODE), ): - ensure_announcement_permission() st = _validate_status(payload.status) ann = Announcement( title=payload.title.strip(), @@ -203,7 +210,6 @@ async def update_announcement( db: AsyncSession = Depends(get_db), _: Dict[str, Any] = require_permission(PERM_CODE), ): - ensure_announcement_permission() ann = await _get_or_404(db, aid) if payload.title is not None: @@ -229,7 +235,6 @@ async def publish_announcement( db: AsyncSession = Depends(get_db), _: Dict[str, Any] = require_permission(PERM_CODE), ): - ensure_announcement_permission() ann = await _get_or_404(db, aid) ann.status = AnnouncementStatus.PUBLISHED if ann.published_at is None: @@ -245,7 +250,6 @@ async def offline_announcement( db: AsyncSession = Depends(get_db), _: Dict[str, Any] = require_permission(PERM_CODE), ): - ensure_announcement_permission() ann = await _get_or_404(db, aid) ann.status = AnnouncementStatus.OFFLINE await db.commit() @@ -259,7 +263,6 @@ async def delete_announcement( db: AsyncSession = Depends(get_db), _: Dict[str, Any] = require_permission(PERM_CODE), ): - ensure_announcement_permission() ann = await _get_or_404(db, aid) await db.delete(ann) await db.commit() diff --git a/backend/app/modules/admin/api/permissions.py b/backend/app/modules/admin/api/permissions.py index 360c0db3..fe1f25e4 100644 --- a/backend/app/modules/admin/api/permissions.py +++ b/backend/app/modules/admin/api/permissions.py @@ -59,6 +59,10 @@ async def get_all_permissions( ensure_task_policy_permission() from app.modules.admin.services.knowledge_permissions import ensure_knowledge_contribution_permission ensure_knowledge_contribution_permission() + # 公告管理:码在 permissions 表里存在才能在权限管理页勾给非超管角色。 + # 必须放在这里(读权限列表时),不能放接口函数体内,否则 403 会先挡住注册。 + from app.modules.admin.api.announcement import ensure_announcement_permission + ensure_announcement_permission() permissions = db_manager.get_all_permissions() return DataResponse( code=0, diff --git a/backend/app/services/permission_service.py b/backend/app/services/permission_service.py index e9ffb314..6932e42c 100644 --- a/backend/app/services/permission_service.py +++ b/backend/app/services/permission_service.py @@ -24,11 +24,21 @@ PERM_PROJECT_LEDGER_IMPORT = "frontend:admin:project-ledger-import:show" PERM_KNOWLEDGE_CONTRIBUTION = "frontend:admin:knowledge:show" +# 后台「公告管理」(活动中心公告的发布 / 下线 / 置顶)的可见/可调判据。 +# 公告是运营动作,默认只开给全局角色「超级管理员」(admin 账号照旧直通); +# 其他角色要在权限管理里显式勾选——该码由 announcement.ensure_announcement_permission +# 在 permissions 的注册块里写入,不会在接口函数体内写(见该函数 docstring)。 +PERM_ANNOUNCEMENT = "frontend:admin:announcement:manage" + # 「拥有哪个全局角色 → 额外获得哪些权限码」。开发者 / 超级管理员可编辑详情模板; -# 一键导入全部项目只开给超级管理员(admin 账号照旧直通)。 +# 一键导入全部项目、公告管理只开给超级管理员(admin 账号照旧直通)。 _GLOBAL_ROLE_DERIVED_PERMISSIONS: Dict[str, tuple] = { "开发者": (PERM_PROJECT_INFO_TEMPLATE,), - "超级管理员": (PERM_PROJECT_INFO_TEMPLATE, PERM_PROJECT_LEDGER_IMPORT), + "超级管理员": ( + PERM_PROJECT_INFO_TEMPLATE, + PERM_PROJECT_LEDGER_IMPORT, + PERM_ANNOUNCEMENT, + ), } _user_roles_cache: Dict[str, Dict[str, Any]] = {} _user_roles_cache_ts: Dict[str, float] = {} diff --git a/backend/tests/test_announcement_api.py b/backend/tests/test_announcement_api.py index a158d5f4..047bfa69 100644 --- a/backend/tests/test_announcement_api.py +++ b/backend/tests/test_announcement_api.py @@ -12,7 +12,8 @@ - POST /api/admin/announcements/{id}/publish | /offline 一律不连真库:conftest 已把 app.core.database 换成 MagicMock;这里 override -get_db 返回可编排的 Mock,并打桩 permissions 表写入(ensure_announcement_permission)。 +get_db 返回可编排的 Mock。权限码注册只应发生在 permissions.get_all_permissions +的注册块里——接口路径里写 permissions 表会直接让本文件用例失败(回归钉)。 """ from datetime import datetime from types import SimpleNamespace @@ -71,9 +72,19 @@ def _ann(**over): @pytest.fixture(autouse=True) -def _stub_permission_write(monkeypatch): - """ensure_announcement_permission 会写 permissions 表,单测里打桩。""" - monkeypatch.setattr(IdentityService, "add_permission", lambda *a, **kw: True) +def _forbid_inline_permission_write(monkeypatch): + """回归钉:接口路径不允许写 permissions 表。 + + 原实现把 ensure_announcement_permission() 放在各后台接口函数体首行,而 + require_permission 是依赖、先于函数体执行 → 没有该权限的账号永远 403, + 注册那行永远走不到(生产自锁)。这里把它设成硬失败,防止有人再加回去。 + """ + def _boom(*args, **kwargs): + raise AssertionError( + "接口路径不应注册权限码(应在 permissions.get_all_permissions 的注册块里)" + ) + + monkeypatch.setattr(IdentityService, "add_permission", _boom) @pytest.fixture @@ -234,3 +245,49 @@ def test_delete_returns_id(make_client, db): assert body == {"ok": True, "id": 1} db.delete.assert_awaited_once() + + +def test_admin_api_allowed_with_derived_permission(make_client, db): + """超管不是靠 admin 账号直通,而是登录态 permissions 里带了这个码。""" + db.execute = AsyncMock(side_effect=[_Res(scalar=0), _Res(rows=[])]) + super_admin = { + "id": "u-sa", + "username": "bob", + "permissions": [ann_api.PERM_CODE], + "roles": {}, + } + + response = make_client(super_admin).get("/api/admin/announcements") + + assert response.status_code == 200 + + +# --------------------------------------------------------------------------- +# 权限码注册:只能走 permissions 的注册块,且超管默认可见 +# --------------------------------------------------------------------------- + +def test_super_admin_role_derives_announcement_permission(): + """全局角色「超级管理员」默认拿到公告管理码(否则后台入口对超管不可见)。""" + from app.services.permission_service import ( + _GLOBAL_ROLE_DERIVED_PERMISSIONS, + PERM_ANNOUNCEMENT, + ) + + assert PERM_ANNOUNCEMENT == ann_api.PERM_CODE + assert PERM_ANNOUNCEMENT in _GLOBAL_ROLE_DERIVED_PERMISSIONS["超级管理员"] + + +def test_ensure_registers_code_for_permission_management(monkeypatch): + """权限管理页要能勾到该码:注册时写入的 code 必须与接口闸门一致。""" + captured = {} + + def _capture(**kwargs): + captured.update(kwargs) + return True + + monkeypatch.setattr(IdentityService, "add_permission", _capture) + + ann_api.ensure_announcement_permission() + + assert captured["code"] == ann_api.PERM_CODE + assert captured["permission_id"] == "perm_frontend_admin_announcement_manage"