From c3a5dc2a4178072acaf01f63ef23500d067328a2 Mon Sep 17 00:00:00 2001 From: no_gen_mu Date: Wed, 8 Jul 2026 19:44:03 +0800 Subject: [PATCH] =?UTF-8?q?=E5=90=8E=E5=8F=B0=E6=9D=83=E9=99=90=E7=AE=A1?= =?UTF-8?q?=E7=90=86=E6=94=AF=E6=8C=81=E3=80=8C=E8=87=AA=E5=AE=9A=E4=B9=89?= =?UTF-8?q?=E3=80=8D=E8=A7=92=E8=89=B2=EF=BC=9A=E6=8C=89=E4=BA=BA=E5=AD=98?= =?UTF-8?q?=20pages=5Foverride=EF=BC=8C=E5=8F=AF=E8=A7=81=E9=A1=B5?= =?UTF-8?q?=E9=80=90=E9=A1=B5=E5=8B=BE=E9=80=89=E5=B9=B6=E5=8D=B3=E6=97=B6?= =?UTF-8?q?=E7=94=9F=E6=95=88?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit admin_user 加 pages_override(可空 JSON)+ Alembic 迁移;role=="custom" 时可见页由这份逐页 勾选决定,否则跟随角色。create/update 收 pages_override(切回普通角色自动清空),_validate_role 放行 custom 哨兵角色。登录/me 下发有效页时优先用 override → 左侧导航直接按勾选生效。 补 4 个用例覆盖建/改/切回/切入,test_admin_roles 全绿。 Co-Authored-By: Claude Opus 4.8 --- alembic/versions/admin_user_pages_override.py | 35 +++++++++ app/admin/permissions.py | 3 + app/admin/repositories/admin_user.py | 5 +- app/admin/routers/admins.py | 33 ++++++-- app/admin/routers/auth.py | 9 ++- app/admin/schemas/admin.py | 6 +- app/admin/schemas/auth.py | 5 +- app/models/admin.py | 5 +- tests/test_admin_roles.py | 77 +++++++++++++++++++ 9 files changed, 167 insertions(+), 11 deletions(-) create mode 100644 alembic/versions/admin_user_pages_override.py diff --git a/alembic/versions/admin_user_pages_override.py b/alembic/versions/admin_user_pages_override.py new file mode 100644 index 0000000..ca49110 --- /dev/null +++ b/alembic/versions/admin_user_pages_override.py @@ -0,0 +1,35 @@ +"""admin_user 加 pages_override 列(「自定义」权限:按人存专属可见页 key 列表) + +权限管理页新增「自定义」角色:选它时该成员的可见页不跟随任何共享角色,而由逐页勾选决定, +存这个人专属的一份页 key 列表。仅 role == "custom" 时有效;普通角色为 None(可见页跟随角色)。 +PG 用 JSONB,SQLite 退化为通用 JSON(同 admin_audit_log.detail / comparison_record.raw_payload)。 + +Revision ID: admin_user_pages_override +Revises: admin_user_plain_password +Create Date: 2026-07-08 00:00:00.000000 +""" + +from collections.abc import Sequence + +import sqlalchemy as sa +from sqlalchemy.dialects.postgresql import JSONB + +from alembic import op + +revision: str = "admin_user_pages_override" +down_revision: str | Sequence[str] | None = "admin_user_plain_password" +branch_labels: str | Sequence[str] | None = None +depends_on: str | Sequence[str] | None = None + +# 与 app/models/admin.py 的 _JSON 一致:PG JSONB / 其它 JSON +_JSON = sa.JSON().with_variant(JSONB(), "postgresql") + + +def upgrade() -> None: + with op.batch_alter_table("admin_user", schema=None) as batch_op: + batch_op.add_column(sa.Column("pages_override", _JSON, nullable=True)) + + +def downgrade() -> None: + with op.batch_alter_table("admin_user", schema=None) as batch_op: + batch_op.drop_column("pages_override") diff --git a/app/admin/permissions.py b/app/admin/permissions.py index e17f7db..ca0ee33 100644 --- a/app/admin/permissions.py +++ b/app/admin/permissions.py @@ -9,6 +9,9 @@ super_admin 为内建全权角色,恒可见全部页(effective_pages 特判)。 from __future__ import annotations SUPER_ADMIN_ROLE = "super_admin" +# 「自定义」哨兵角色:不是 admin_role 表里的行,而是标记「这个人的可见页由 pages_override 决定」。 +# admin_user.role == CUSTOM_ROLE 时,有效可见页取 admin_user.pages_override(见 auth._admin_out_with_pages)。 +CUSTOM_ROLE = "custom" # 分组镜像前端导航(app/(main)/layout.tsx 的 NAV_GROUPS);key = 路由一级 PERMISSION_CATALOG: list[dict] = [ diff --git a/app/admin/repositories/admin_user.py b/app/admin/repositories/admin_user.py index 828efe7..d9055df 100644 --- a/app/admin/repositories/admin_user.py +++ b/app/admin/repositories/admin_user.py @@ -26,14 +26,17 @@ def create_admin( password: str, role: str = "operator", plain_password: str | None = None, + pages_override: list[str] | None = None, ) -> AdminUser: """建管理员。plain_password 非空则额外留存明文(后台 UI 建的账号传,供权限管理页复看); - 脚本/起后台建账号不传(留 None → 前端不显示密码)。""" + 脚本/起后台建账号不传(留 None → 前端不显示密码)。 + pages_override:role=="custom" 时传专属可见页 key 列表;普通角色为 None。""" admin = AdminUser( username=username, password_hash=hash_password(password), role=role, plain_password=plain_password, + pages_override=pages_override, ) db.add(admin) db.commit() diff --git a/app/admin/routers/admins.py b/app/admin/routers/admins.py index 5273a45..60d9511 100644 --- a/app/admin/routers/admins.py +++ b/app/admin/routers/admins.py @@ -5,7 +5,7 @@ from fastapi import APIRouter, Depends, HTTPException, Request from app.admin.audit import write_audit from app.admin.deps import AdminDb, CurrentAdmin, get_client_ip, require_role -from app.admin.permissions import SUPER_ADMIN_ROLE +from app.admin.permissions import CUSTOM_ROLE, SUPER_ADMIN_ROLE, sanitize_pages from app.admin.repositories import admin_role as role_repo from app.admin.repositories import admin_user as admin_repo from app.admin.schemas.admin import AdminCreateRequest, AdminUpdateRequest @@ -26,14 +26,22 @@ def _active_super_count(db: AdminDb) -> int: def _validate_role(db: AdminDb, role: str) -> None: - """角色必须是 super_admin 或 admin_role 表里已存在的角色,否则 400。""" - if role == SUPER_ADMIN_ROLE: + """角色必须是 super_admin / custom(自定义) / admin_role 表里已存在的角色,否则 400。""" + if role in (SUPER_ADMIN_ROLE, CUSTOM_ROLE): return role_repo.ensure_builtin_roles(db) # 空表(测试/全新库)兜底播种,再校验 if role_repo.get_role(db, role) is None: raise HTTPException(status_code=400, detail=f"角色不存在: {role}") +def _clean_override(role: str, pages_override: list[str] | None) -> list[str] | None: + """按最终角色算出该存的 pages_override:custom → 勾选集(过滤非法 key,可空列表); + 非 custom → None(切回普通角色即清空自定义页)。""" + if role == CUSTOM_ROLE: + return sanitize_pages(pages_override) + return None + + @router.get("", response_model=list[AdminOut], summary="管理员列表(含明文密码,super_admin 专属)") def list_admins(db: AdminDb) -> list[AdminOut]: out: list[AdminOut] = [] @@ -51,13 +59,16 @@ def create_admin( if admin_repo.get_by_username(db, body.username) is not None: raise HTTPException(status_code=409, detail="用户名已存在") _validate_role(db, body.role) + override = _clean_override(body.role, body.pages_override) new = admin_repo.create_admin( db, username=body.username, password=body.password, role=body.role, plain_password=body.password, # UI 建的账号留存明文,供权限管理页复看 + pages_override=override, ) write_audit( db, admin, action="admin.create", target_type="admin", target_id=new.id, - detail={"username": new.username, "role": new.role}, ip=get_client_ip(request), commit=True, + detail={"username": new.username, "role": new.role, "pages_override": override}, + ip=get_client_ip(request), commit=True, ) return AdminOut.model_validate(new) @@ -89,7 +100,8 @@ def update_admin( _validate_role(db, body.role) changes: dict = {} - if body.role is not None and body.role != target.role: + role_changed = body.role is not None and body.role != target.role + if role_changed: changes["role"] = {"before": target.role, "after": body.role} target.role = body.role if body.status is not None and body.status != target.status: @@ -99,6 +111,17 @@ def update_admin( changes["password"] = "reset" target.password_hash = hash_password(body.password) target.plain_password = body.password # 同步留存明文,权限管理页复看保持一致 + + # 自定义可见页:按「最终角色」(target.role,已应用完角色变更)决定该存什么。 + # - 切到/维持 custom 且传了 pages_override → 用勾选集;切到 custom 没传 → 清成空列表。 + # - 切回普通角色 → 清空 override(_clean_override 返回 None)。 + # - 角色没变、只传 pages_override(编辑现有 custom 用户的勾选)→ 也更新。 + if role_changed or body.pages_override is not None: + new_override = _clean_override(target.role, body.pages_override) + if new_override != target.pages_override: + changes["pages_override"] = {"before": target.pages_override, "after": new_override} + target.pages_override = new_override + if not changes: raise HTTPException(status_code=400, detail="无任何变更字段") db.commit() diff --git a/app/admin/routers/auth.py b/app/admin/routers/auth.py index 6aa8e35..27b5bb3 100644 --- a/app/admin/routers/auth.py +++ b/app/admin/routers/auth.py @@ -6,6 +6,7 @@ import logging from fastapi import APIRouter, Depends, HTTPException from app.admin.deps import AdminDb, CurrentAdmin +from app.admin.permissions import CUSTOM_ROLE, sanitize_pages from app.admin.repositories import admin_role as role_repo from app.admin.repositories import admin_user as admin_repo from app.admin.schemas.auth import AdminLoginRequest, AdminLoginResponse, AdminOut @@ -19,9 +20,13 @@ router = APIRouter(prefix="/admin/api/auth", tags=["admin-auth"]) def _admin_out_with_pages(admin, db: AdminDb) -> AdminOut: # noqa: ANN001 - """AdminOut + 当前角色有效可见页(前端左侧导航按此过滤)。""" + """AdminOut + 有效可见页(前端左侧导航按此过滤)。 + role=="custom" → 用这个人的 pages_override(按人自定义,过滤悬空 key);其余走角色解析。""" out = AdminOut.model_validate(admin) - out.pages = role_repo.effective_pages_of(db, admin.role) + if admin.role == CUSTOM_ROLE: + out.pages = sanitize_pages(admin.pages_override) + else: + out.pages = role_repo.effective_pages_of(db, admin.role) return out diff --git a/app/admin/schemas/admin.py b/app/admin/schemas/admin.py index fcde8bf..ae12752 100644 --- a/app/admin/schemas/admin.py +++ b/app/admin/schemas/admin.py @@ -13,14 +13,18 @@ class AdminCreateRequest(BaseModel): username: str = Field(..., min_length=3, max_length=64) password: str = Field(..., min_length=8, max_length=72) # bcrypt ≤72 字节 role: str = Field("operator", min_length=1, max_length=32) + # 仅当 role == "custom":这个人专属可见页 key 列表(逐页勾选结果)。其余角色不传/忽略。 + pages_override: list[str] | None = None class AdminUpdateRequest(BaseModel): - """改角色 / 启用禁用 / 重置密码,字段都可选(只改传了的)。""" + """改角色 / 启用禁用 / 重置密码 / 自定义可见页,字段都可选(只改传了的)。""" role: str | None = Field(None, min_length=1, max_length=32) status: Literal["active", "disabled"] | None = None password: str | None = Field(None, min_length=8, max_length=72) + # 改成/更新「自定义」可见页;role 切回普通角色时后端会清空 override(见路由)。 + pages_override: list[str] | None = None class AdminAuditLogOut(BaseModel): diff --git a/app/admin/schemas/auth.py b/app/admin/schemas/auth.py index 3972fed..ef64114 100644 --- a/app/admin/schemas/auth.py +++ b/app/admin/schemas/auth.py @@ -21,8 +21,11 @@ class AdminOut(BaseModel): created_at: datetime last_login_at: datetime | None = None # 该管理员当前角色的有效可见页(= 左侧导航项 key);仅登录 / /me 填充,列表接口默认空。 - # 前端据此过滤左侧导航(super_admin = 全部页)。见 app/admin/permissions.py。 + # 前端据此过滤左侧导航(super_admin = 全部页;role=="custom" = pages_override)。见 permissions.py。 pages: list[str] = [] + # 「自定义」可见页原始勾选集(role=="custom" 时非空);列表接口下发,供权限管理页编辑回填勾选。 + # 普通角色为 None。from_attributes 自动从 ORM 列取。 + pages_override: list[str] | None = None # 明文登录密码:仅「管理员账号列表」(super_admin 专属路由)填充,供权限管理页编辑时复看; # 无留存(脚本建的超管 / 旧账号)为 None → 前端不显示。登录 / /me 不下发(保持 None)。 password: str | None = None diff --git a/app/models/admin.py b/app/models/admin.py index 0c56963..8bb72b4 100644 --- a/app/models/admin.py +++ b/app/models/admin.py @@ -28,8 +28,11 @@ class AdminUser(Base): # 明文登录密码:仅「后台 UI 创建/重置」的管理员留存,供超管在权限管理页复看转交。 # 脚本/起后台时建的超管账号不写(为 None → 前端「不显示密码」)。⚠️ 内部工具便利取舍,见 create/list。 plain_password: Mapped[str | None] = mapped_column(String(128), nullable=True) - # super_admin(全权+管账号)/ finance(钱:提现+金币)/ operator(用户+反馈+大盘) + # super_admin(全权+管账号)/ finance(钱:提现+金币)/ operator(用户+反馈+大盘)/ custom(按人自定义) role: Mapped[str] = mapped_column(String(20), nullable=False, default="operator") + # 「自定义」权限:仅当 role == "custom" 时有效,存这个人专属的可见页 key 列表(不共享给他人)。 + # 非 custom 用户为 None → 可见页跟随角色。登录/`/me` 下发有效页时,非空即优先用它(见 auth._admin_out_with_pages)。 + pages_override: Mapped[list | None] = mapped_column(_JSON, nullable=True) # active / disabled status: Mapped[str] = mapped_column(String(20), nullable=False, default="active") diff --git a/tests/test_admin_roles.py b/tests/test_admin_roles.py index ce3f4ca..85ab010 100644 --- a/tests/test_admin_roles.py +++ b/tests/test_admin_roles.py @@ -150,3 +150,80 @@ def test_create_admin_rejects_unknown_role(admin_client, super_token) -> None: headers=_auth(super_token), ) assert r.status_code == 400 + + +def _login_pages(username: str, password: str = "pass1234") -> list[str]: + """以某账号登录,取下发的有效可见页(左侧导航过滤依据)。""" + c = TestClient(admin_app) + return c.post( + "/admin/api/auth/login", json={"username": username, "password": password} + ).json()["admin"]["pages"] + + +def test_custom_role_create_pages_take_effect(admin_client, super_token) -> None: + # 建「自定义」账号:role=custom + 勾选页(含一个非法 key,应被过滤) + r = admin_client.post( + "/admin/api/admins", + json={ + "username": "cust_user", "password": "pass1234", "role": "custom", + "pages_override": ["dashboard", "withdraws", "xx-bad"], + }, + headers=_auth(super_token), + ) + assert r.status_code == 200, r.text + # 登录下发的 pages == 勾选集(非法 key 过滤),真正驱动左侧导航 + assert set(_login_pages("cust_user")) == {"dashboard", "withdraws"} + # 账号列表回显原始勾选集(供编辑回填),同样已过滤 + admins = {a["username"]: a for a in admin_client.get("/admin/api/admins", headers=_auth(super_token)).json()} + assert set(admins["cust_user"]["pages_override"]) == {"dashboard", "withdraws"} + assert admins["cust_user"]["role"] == "custom" + + +def test_custom_role_update_and_switch_back_clears_override(admin_client, super_token) -> None: + admin_client.post( + "/admin/api/admins", + json={ + "username": "cust_sw", "password": "pass1234", "role": "custom", + "pages_override": ["dashboard"], + }, + headers=_auth(super_token), + ) + aid = next( + a["id"] for a in admin_client.get("/admin/api/admins", headers=_auth(super_token)).json() + if a["username"] == "cust_sw" + ) + # 只改勾选集(角色不变)→ 生效 + u = admin_client.patch( + f"/admin/api/admins/{aid}", json={"pages_override": ["dashboard", "feedbacks"]}, + headers=_auth(super_token), + ) + assert u.status_code == 200, u.text + assert set(_login_pages("cust_sw")) == {"dashboard", "feedbacks"} + # 切回普通角色 operator → override 清空,pages 跟随角色(运营页集,且不含 admins) + u2 = admin_client.patch( + f"/admin/api/admins/{aid}", json={"role": "operator"}, headers=_auth(super_token) + ) + assert u2.status_code == 200, u2.text + admins = {a["username"]: a for a in admin_client.get("/admin/api/admins", headers=_auth(super_token)).json()} + assert admins["cust_sw"]["pages_override"] is None + pages = set(_login_pages("cust_sw")) + assert "feedbacks" in pages and "admins" not in pages # 运营口径,自定义页已不生效 + + +def test_switch_operator_to_custom_sets_override(admin_client, super_token) -> None: + admin_client.post( + "/admin/api/admins", + json={"username": "op2cust", "password": "pass1234", "role": "operator"}, + headers=_auth(super_token), + ) + aid = next( + a["id"] for a in admin_client.get("/admin/api/admins", headers=_auth(super_token)).json() + if a["username"] == "op2cust" + ) + u = admin_client.patch( + f"/admin/api/admins/{aid}", + json={"role": "custom", "pages_override": ["config"]}, + headers=_auth(super_token), + ) + assert u.status_code == 200, u.text + assert set(_login_pages("op2cust")) == {"config"} -- 2.52.0