Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
21 changes: 12 additions & 9 deletions backend/app/modules/admin/api/announcement.py
Original file line number Diff line number Diff line change
Expand Up @@ -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。
Expand All @@ -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,
Expand All @@ -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,
Expand Down Expand Up @@ -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))
Expand All @@ -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(),
Expand All @@ -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:
Expand All @@ -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:
Expand All @@ -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()
Expand All @@ -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()
Expand Down
4 changes: 4 additions & 0 deletions backend/app/modules/admin/api/permissions.py
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down
14 changes: 12 additions & 2 deletions backend/app/services/permission_service.py
Original file line number Diff line number Diff line change
Expand Up @@ -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] = {}
Expand Down
65 changes: 61 additions & 4 deletions backend/tests/test_announcement_api.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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"
Loading