tao.chen/model-platform
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: soft-deleted permission readd

Browse Source
This commit is contained in:
tao.chen
2026-09-02 10:10:41 +08:00
committed by tao.chen
parent 7c7e1a99a9
commit 36c402d407
5 changed files with 485 additions and 142 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/pyproject.toml
+1
View File
@@ -31,6 +31,7 @@ build-backend = "hatchling.build"
[dependency-groups]
dev = [
"aiosqlite>=0.22.1",
"pytest>=9.1.1",
"pytest-asyncio>=1.4.0",
"respx>=0.23.1",
backend/src/backend/api/platform/_permission_set.py
+193
View File
@@ -0,0 +1,193 @@
"""Role permission-set replacement logic.
``_apply_role_permission_set`` is shared by PATCH /roles/{role_code}/permissions
and POST /roles so both entry points apply the identical guard order and the
diff-based write. Extracted from ``roles.py`` so the endpoint module stays
under 500 lines.
The write is upsert-style because ``role_permissions`` uses the composite
primary key ``(role_id, permission_id)``: a soft-deleted row (``is_deleted=1``)
still occupies its PK slot, so re-adding a permission must *revive* the
historical row (UPDATE) instead of INSERTing over it (which would raise a
duplicate-key IntegrityError).
"""
from __future__ import annotations
import datetime
from common.db.models import Permissions, RolePermissions, Roles
from fastapi import HTTPException, status
from sqlalchemy import insert, select, update
from sqlalchemy.ext.asyncio import AsyncSession
async def _load_role_permission_codes(
session: AsyncSession, role_id: str
) -> list[str]:
"""Return the active permission_codes for a role, ordered by code."""
rows = (
await session.execute(
select(Permissions.permission_code)
.join(
RolePermissions,
RolePermissions.permission_id == Permissions.permission_id,
)
.where(
RolePermissions.role_id == role_id,
RolePermissions.is_deleted == 0,
Permissions.is_deleted == 0,
)
.order_by(Permissions.permission_code)
)
).all()
return [row[0] for row in rows]
async def _apply_role_permission_set(
session: AsyncSession, role: Roles, codes: list[str]
) -> list[str]:
"""Validate ``codes`` then replace the role's permission set wholesale.
Shared by PATCH /roles/{role_code}/permissions and POST /roles so the
guard order is identical no matter the entry point:
1. Admin: must keep ``system:view`` → 409 (menu perms never gate
API access; auth keys off ``role_code == 'admin'``).
2. Non-admin: ``system.*`` codes → 422.
3. Unknown codes → 422.
4. Diff-based soft-delete + upsert — the ``(role_id, permission_id)``
PK keeps soft-deleted rows, so delete-all/insert-all would
IntegrityError. Repeat-with-same-set is a no-op.
Returns the role's final permission_codes (after flush).
"""
new_codes = list(dict.fromkeys(codes))
if role.role_code == "admin":
keeps_admin_entry = "system:view" in new_codes
if not keeps_admin_entry:
raise HTTPException(
status.HTTP_409_CONFLICT,
"admin 角色必须保留 system:view 权限",
)
else:
# Menu permissions are a frontend-display signal only — backend
# authorization keeps keying off role_code == "admin". Letting a
# non-admin role hold system.* permissions would render the
# system-admin entry in the developer's UI while every
# /api/v1/platform/* call still returns 403. Reject with 422 so
# the failure is unambiguous about *what* the input violated.
leaked_system = [
code for code in new_codes if code.startswith("system:")
]
if leaked_system:
raise HTTPException(
status.HTTP_422_UNPROCESSABLE_ENTITY,
f"非 admin 角色不能拥有 system.* 权限: {leaked_system}",
)
# 3. Validate every requested permission_code exists and is live.
if new_codes:
rows = (
await session.execute(
select(Permissions.permission_code).where(
Permissions.permission_code.in_(new_codes),
Permissions.is_deleted == 0,
)
)
).all()
found = {row[0] for row in rows}
missing = [code for code in new_codes if code not in found]
if missing:
raise HTTPException(
status.HTTP_422_UNPROCESSABLE_ENTITY,
f"未知的 permission_code: {missing}",
)
# 4. Write: diff-based soft-delete + upsert.
# The (role_id, permission_id) PRIMARY KEY still occupies the slot
# of soft-deleted rows, so a "delete-all then insert-all" approach
# would IntegrityError on any code that was already linked.
# Instead: only soft-delete codes NOT in the new set, only revive-or-
# insert codes NOT already active. Repeat-with-same-payload is a no-op.
now = datetime.datetime.utcnow()
current_codes = set(
await _load_role_permission_codes(session, role.role_id)
)
new_set = set(new_codes)
codes_to_drop = current_codes - new_set
codes_to_add = new_set - current_codes
if codes_to_drop:
# Resolve to permission_ids then soft-delete by id pair.
drop_ids = (
await session.execute(
select(Permissions.permission_id).where(
Permissions.permission_code.in_(codes_to_drop),
Permissions.is_deleted == 0,
)
)
).all()
drop_id_values = [row[0] for row in drop_ids]
await session.execute(
update(RolePermissions)
.where(
RolePermissions.role_id == role.role_id,
RolePermissions.permission_id.in_(drop_id_values),
RolePermissions.is_deleted == 0,
)
.values(is_deleted=1, deleted_at=now)
)
if codes_to_add:
add_ids = (
await session.execute(
select(Permissions.permission_id).where(
Permissions.permission_code.in_(codes_to_add),
Permissions.is_deleted == 0,
)
)
).all()
add_id_values = [pid for pid, in add_ids]
if not add_id_values:
return
# 区分「历史软删行」(复活) vs 「全新行」(插入)
# RolePermissions 的主键 (role_id, permission_id) 即使 is_deleted=1 也占槽,
# 直接 INSERT 会撞 PK。复活 + 插入两步走。
existing_ids = set(
(
await session.execute(
select(RolePermissions.permission_id).where(
RolePermissions.role_id == role.role_id,
RolePermissions.permission_id.in_(add_id_values),
)
)
).scalars()
)
revive_ids = [pid for pid in add_id_values if pid in existing_ids]
fresh_ids = [pid for pid in add_id_values if pid not in existing_ids]
if revive_ids:
await session.execute(
update(RolePermissions)
.where(
RolePermissions.role_id == role.role_id,
RolePermissions.permission_id.in_(revive_ids),
)
.values(is_deleted=0, deleted_at=None)
)
if fresh_ids:
await session.execute(
insert(RolePermissions),
[
{"role_id": role.role_id, "permission_id": pid}
for pid in fresh_ids
],
)
await session.flush()
return await _load_role_permission_codes(session, role.role_id)
backend/src/backend/api/platform/roles.py
+7 -142
View File
@@ -1,6 +1,6 @@
"""Platform role & permission management endpoints.
Seven endpoints: role list/CRUD, per-role permission get/patch, and the permission
catalog; ``_apply_role_permission_set`` is shared by PATCH /roles/{code}/permissions and POST /roles.
catalog. The shared permission-set writer lives in :mod:`backend.api.platform._permission_set`.
"""
from __future__ import annotations
@@ -11,7 +11,6 @@ from typing import Any
from common.db.models import (
Permissions,
RolePermissions,
Roles,
Users,
WorkspaceMembers,
@@ -19,7 +18,7 @@ from common.db.models import (
from common.ids import new_ulid
from fastapi import APIRouter, Depends, HTTPException, status
from pydantic import BaseModel, ConfigDict, Field, field_validator
from sqlalchemy import func, insert, select, update
from sqlalchemy import func, select
from sqlalchemy.ext.asyncio import AsyncSession
from backend.api.dependencies import database_session
@@ -29,6 +28,11 @@ from backend.api.platform._deps import (
system_admin_context,
)
from backend.api.platform._permission_set import (
_apply_role_permission_set,
_load_role_permission_codes,
)
router = APIRouter(prefix="/api/v1/platform", tags=["platform"])
RESERVED_PLATFORM_ROLE_CODES = frozenset({"admin", "developer"})
@@ -101,27 +105,6 @@ async def _load_platform_role_by_code(
)
return role
async def _load_role_permission_codes(
session: AsyncSession, role_id: str
) -> list[str]:
"""Return the active permission_codes for a role, ordered by code."""
rows = (
await session.execute(
select(Permissions.permission_code)
.join(
RolePermissions,
RolePermissions.permission_id == Permissions.permission_id,
)
.where(
RolePermissions.role_id == role_id,
RolePermissions.is_deleted == 0,
Permissions.is_deleted == 0,
)
.order_by(Permissions.permission_code)
)
).all()
return [row[0] for row in rows]
def _role_payload(role: Roles, permission_codes: list[str]) -> dict[str, Any]:
return {
"role_id": role.role_id,
@@ -167,124 +150,6 @@ async def get_role_permissions(
context.request_id, _role_payload(role, codes),
)
async def _apply_role_permission_set(
session: AsyncSession, role: Roles, codes: list[str]
) -> list[str]:
"""Validate ``codes`` then replace the role's permission set wholesale.
Shared by PATCH /roles/{role_code}/permissions and POST /roles so the
guard order is identical no matter the entry point:
1. Admin: must keep ``system:view`` → 409 (menu perms never gate
API access; auth keys off ``role_code == 'admin'``).
2. Non-admin: ``system.*`` codes → 422.
3. Unknown codes → 422.
4. Diff-based soft-delete + insert — the ``(role_id, permission_id)``
PK keeps soft-deleted rows, so delete-all/insert-all would
IntegrityError. Repeat-with-same-set is a no-op.
Returns the role's final permission_codes (after flush).
"""
new_codes = list(dict.fromkeys(codes))
if role.role_code == "admin":
keeps_admin_entry = "system:view" in new_codes
if not keeps_admin_entry:
raise HTTPException(
status.HTTP_409_CONFLICT,
"admin 角色必须保留 system:view 权限",
)
else:
# Menu permissions are a frontend-display signal only — backend
# authorization keeps keying off role_code == "admin". Letting a
# non-admin role hold system.* permissions would render the
# system-admin entry in the developer's UI while every
# /api/v1/platform/* call still returns 403. Reject with 422 so
# the failure is unambiguous about *what* the input violated.
leaked_system = [
code for code in new_codes if code.startswith("system:")
]
if leaked_system:
raise HTTPException(
status.HTTP_422_UNPROCESSABLE_ENTITY,
f"非 admin 角色不能拥有 system.* 权限: {leaked_system}",
)
# 3. Validate every requested permission_code exists and is live.
if new_codes:
rows = (
await session.execute(
select(Permissions.permission_code).where(
Permissions.permission_code.in_(new_codes),
Permissions.is_deleted == 0,
)
)
).all()
found = {row[0] for row in rows}
missing = [code for code in new_codes if code not in found]
if missing:
raise HTTPException(
status.HTTP_422_UNPROCESSABLE_ENTITY,
f"未知的 permission_code: {missing}",
)
# 4. Write: diff-based soft-delete + insert.
# The (role_id, permission_id) PRIMARY KEY still occupies the slot
# of soft-deleted rows, so a "delete-all then insert-all" approach
# would IntegrityError on any code that was already linked.
# Instead: only soft-delete codes NOT in the new set, only INSERT
# codes NOT already active. Repeat-with-same-payload is a no-op.
now = datetime.datetime.utcnow()
current_codes = set(
await _load_role_permission_codes(session, role.role_id)
)
new_set = set(new_codes)
codes_to_drop = current_codes - new_set
codes_to_add = new_set - current_codes
if codes_to_drop:
# Resolve to permission_ids then soft-delete by id pair.
drop_ids = (
await session.execute(
select(Permissions.permission_id).where(
Permissions.permission_code.in_(codes_to_drop),
Permissions.is_deleted == 0,
)
)
).all()
drop_id_values = [row[0] for row in drop_ids]
await session.execute(
update(RolePermissions)
.where(
RolePermissions.role_id == role.role_id,
RolePermissions.permission_id.in_(drop_id_values),
RolePermissions.is_deleted == 0,
)
.values(is_deleted=1, deleted_at=now)
)
if codes_to_add:
add_ids = (
await session.execute(
select(Permissions.permission_id).where(
Permissions.permission_code.in_(codes_to_add),
Permissions.is_deleted == 0,
)
)
).all()
if add_ids:
await session.execute(
insert(RolePermissions),
[
{"role_id": role.role_id, "permission_id": pid}
for pid, in add_ids
],
)
await session.flush()
return await _load_role_permission_codes(session, role.role_id)
@router.patch("/roles/{role_code}/permissions")
async def patch_role_permissions(
role_code: str,
backend/tests/conftest.py
+32
View File
@@ -0,0 +1,32 @@
"""Shared test bootstrap.
Loads the repo-root ``.env`` so modules that import ``common.config``
(which is a process-wide singleton) can resolve ``APP_CONFIG_SECRET_KEY``
at collection time. Without it, any test importing ``backend.api.*`` fails
with a RuntimeError about ENC(...) ciphertext.
"""
from __future__ import annotations
import os
from pathlib import Path
_REPO_ROOT = Path(__file__).resolve().parent.parent.parent
def _bootstrap() -> None:
if os.environ.get("APP_CONFIG_SECRET_KEY"):
return
env_file = _REPO_ROOT / ".env"
if not env_file.is_file():
return
try:
from dotenv import dotenv_values
except ImportError:
return
value = dotenv_values(env_file).get("APP_CONFIG_SECRET_KEY")
if value:
os.environ["APP_CONFIG_SECRET_KEY"] = value
_bootstrap()
backend/tests/test_role_permissions_re_add.py
+252
View File
@@ -0,0 +1,252 @@
"""Regression tests for the ``role_permissions`` re-add (upsert) flow.
Reproduces the PATCH /api/v1/platform/roles/admin/permissions bug where
re-adding a permission that was previously soft-deleted collides with the
composite ``(role_id, permission_id)`` PRIMARY KEY: the soft-deleted row
still occupies its PK slot, so a plain INSERT raises a duplicate-key
IntegrityError (MySQL 1062).
Uses sqlite in-memory + aiosqlite (``create_async_engine``) and drives
:func:`_apply_role_permission_set` directly — no HTTP layer, no MySQL.
"""
from __future__ import annotations
import datetime
import pytest
from sqlalchemy import func, insert, select, text
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
from sqlalchemy.pool import StaticPool
from common.db.models import Permissions, RolePermissions, Roles
from backend.api.platform._permission_set import _apply_role_permission_set
# -- realistic admin permission codes (matches the reported repro) ----------
SYSTEM_VIEW = "system:view"
SYSTEM_USER_VIEW = "system:user:view"
SYSTEM_PROJECT_VIEW = "system:project:view"
# Raw DDL mirrors the ORM models (roles/permissions/role_permissions) with
# sqlite-compatible types — the MySQL dialects (CHAR/TINYINT/DATETIME(fsp))
# cannot be compiled by SQLite's DDL compiler, so the tables are created by
# hand and then driven through the ORM at runtime.
_DDL = [
"""
CREATE TABLE roles (
role_id VARCHAR(26) PRIMARY KEY,
role_code VARCHAR(64) NOT NULL UNIQUE,
role_name VARCHAR(100) NOT NULL,
role_scope VARCHAR(16) NOT NULL,
is_builtin INTEGER NOT NULL DEFAULT 0,
created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
description VARCHAR(500),
is_deleted INTEGER NOT NULL DEFAULT 0,
deleted_at DATETIME
)
""",
"""
CREATE TABLE permissions (
permission_id VARCHAR(26) PRIMARY KEY,
permission_code VARCHAR(128) NOT NULL UNIQUE,
permission_name VARCHAR(100) NOT NULL,
module_code VARCHAR(64) NOT NULL,
created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
description VARCHAR(500),
is_deleted INTEGER NOT NULL DEFAULT 0,
deleted_at DATETIME
)
""",
"""
CREATE TABLE role_permissions (
role_id VARCHAR(26) NOT NULL,
permission_id VARCHAR(26) NOT NULL,
created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
is_deleted INTEGER NOT NULL DEFAULT 0,
deleted_at DATETIME,
PRIMARY KEY (role_id, permission_id)
)
""",
]
@pytest.fixture
async def session():
"""Async in-memory sqlite session with the three identity tables."""
engine = create_async_engine(
"sqlite+aiosqlite://",
poolclass=StaticPool,
connect_args={"check_same_thread": False},
)
async with engine.begin() as conn:
for ddl in _DDL:
await conn.execute(text(ddl))
Session = async_sessionmaker(engine, expire_on_commit=False)
async with Session() as db:
yield db
await engine.dispose()
# -- helpers ---------------------------------------------------------------
async def _add_permission(session, permission_id: str, code: str) -> None:
session.add(
Permissions(
permission_id=permission_id,
permission_code=code,
permission_name=code,
module_code="system",
description=None,
)
)
await session.flush()
async def _add_role(session, role_id: str, role_code: str) -> Roles:
role = Roles(
role_id=role_id,
role_code=role_code,
role_name=role_code,
role_scope="platform",
is_builtin=1,
description=None,
)
session.add(role)
await session.flush()
return role
async def _link(session, role_id: str, permission_id: str, *, active: bool) -> None:
"""Insert (or directly mark) a role_permissions row with a given state."""
await session.execute(
insert(RolePermissions),
[
{
"role_id": role_id,
"permission_id": permission_id,
"is_deleted": 0 if active else 1,
"deleted_at": None if active else datetime.datetime(2026, 1, 1),
}
],
)
await session.flush()
async def _active_count(session, role_id: str) -> int:
return int(
await session.scalar(
select(func.count())
.select_from(RolePermissions)
.where(
RolePermissions.role_id == role_id,
RolePermissions.is_deleted == 0,
)
)
or 0
)
async def _active_row_count(session, role_id: str) -> int:
return int(
await session.scalar(
select(func.count())
.select_from(RolePermissions)
.where(RolePermissions.role_id == role_id)
)
or 0
)
# -- tests -----------------------------------------------------------------
async def test_happy_path_soft_delete_then_readd(session) -> None:
"""Soft-delete two permissions, then re-add them — no IntegrityError."""
role = await _add_role(session, "R" * 26, "admin")
await _add_permission(session, "P1" * 13, SYSTEM_VIEW)
await _add_permission(session, "P2" * 13, SYSTEM_USER_VIEW)
await _add_permission(session, "P3" * 13, SYSTEM_PROJECT_VIEW)
view_id = "P1" * 13
user_view_id = "P2" * 13
project_view_id = "P3" * 13
for pid in (view_id, user_view_id, project_view_id):
await _link(session, role.role_id, pid, active=True)
# 1st PATCH: shrink to only system:view → the other two get soft-deleted.
codes = await _apply_role_permission_set(session, role, [SYSTEM_VIEW])
assert codes == [SYSTEM_VIEW]
assert await _active_count(session, role.role_id) == 1
assert await _active_row_count(session, role.role_id) == 3
# 2nd PATCH: restore the full set → the soft-deleted rows must be
# revived (UPDATE), not INSERTed over (would raise IntegrityError).
codes = await _apply_role_permission_set(
session,
role,
[SYSTEM_VIEW, SYSTEM_USER_VIEW, SYSTEM_PROJECT_VIEW],
)
assert codes == [SYSTEM_PROJECT_VIEW, SYSTEM_USER_VIEW, SYSTEM_VIEW]
assert await _active_count(session, role.role_id) == 3
assert await _active_row_count(session, role.role_id) == 3
states = (
await session.execute(
select(RolePermissions.is_deleted, RolePermissions.deleted_at).where(
RolePermissions.role_id == role.role_id
)
)
).all()
assert all(is_deleted == 0 and deleted_at is None for is_deleted, deleted_at in states)
async def test_fresh_insert_no_history(session) -> None:
"""Brand-new role: a never-before-linked permission is plain INSERTed."""
role = await _add_role(session, "R" * 26, "admin")
await _add_permission(session, "P1" * 13, SYSTEM_VIEW)
codes = await _apply_role_permission_set(session, role, [SYSTEM_VIEW])
assert codes == [SYSTEM_VIEW]
assert await _active_count(session, role.role_id) == 1
assert await _active_row_count(session, role.role_id) == 1
async def test_mixed_revive_and_fresh_insert(session) -> None:
"""One historically soft-deleted row is revived, one fresh row inserted."""
role = await _add_role(session, "R" * 26, "admin")
await _add_permission(session, "P1" * 13, SYSTEM_VIEW)
await _add_permission(session, "P2" * 13, SYSTEM_USER_VIEW)
await _link(session, role.role_id, "P1" * 13, active=False) # 历史软删行
codes = await _apply_role_permission_set(
session, role, [SYSTEM_VIEW, SYSTEM_USER_VIEW]
)
assert codes == [SYSTEM_USER_VIEW, SYSTEM_VIEW]
assert await _active_count(session, role.role_id) == 2
assert await _active_row_count(session, role.role_id) == 2
revived = (
await session.execute(
select(RolePermissions.is_deleted, RolePermissions.deleted_at).where(
RolePermissions.role_id == role.role_id,
RolePermissions.permission_id == "P1" * 13,
)
)
).one()
assert revived[0] == 0 and revived[1] is None
async def test_same_set_is_noop(session) -> None:
"""PATCHing the exact same active set again changes nothing."""
role = await _add_role(session, "R" * 26, "admin")
await _add_permission(session, "P1" * 13, SYSTEM_VIEW)
await _link(session, role.role_id, "P1" * 13, active=True)
first = await _apply_role_permission_set(session, role, [SYSTEM_VIEW])
second = await _apply_role_permission_set(session, role, [SYSTEM_VIEW])
assert first == [SYSTEM_VIEW]
assert second == [SYSTEM_VIEW]
assert await _active_count(session, role.role_id) == 1
assert await _active_row_count(session, role.role_id) == 1